/usr/lib/python3/dist-packages/maascli/config.py is in python3-maas-client 2.4.0~beta2-6865-gec43e47e6-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | # Copyright 2012-2016 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Configuration abstractions for the MAAS CLI."""
__all__ = [
"ProfileConfig",
]
from contextlib import (
closing,
contextmanager,
)
import json
import os
from os.path import expanduser
import sqlite3
from maascli import utils
class ProfileConfig:
"""Store profile configurations in an sqlite3 database."""
def __init__(self, database):
self.database = database
self.cache = {}
with self.cursor() as cursor:
cursor.execute(
"CREATE TABLE IF NOT EXISTS profiles "
"(id INTEGER PRIMARY KEY,"
" name TEXT NOT NULL UNIQUE,"
" data BLOB)")
self.__fill_cache()
def cursor(self):
return closing(self.database.cursor())
def __fill_cache(self):
"""Touch each entry in the database to fill the cache. This cache is
needed to enforce a consistent view. Without it, the list of items can
be out of sync with the items actually in the database leading to
KeyErrors when traversing the profiles.
"""
for name in self:
try:
self[name]
except KeyError:
pass
def __iter__(self):
if self.cache:
return (name for name in self.cache)
with self.cursor() as cursor:
results = cursor.execute(
"SELECT name FROM profiles").fetchall()
return (name for (name,) in results)
def __getitem__(self, name):
if name in self.cache:
return self.cache[name]
with self.cursor() as cursor:
data = cursor.execute(
"SELECT data FROM profiles"
" WHERE name = ?", (name,)).fetchone()
if data is None:
raise KeyError(name)
else:
info = json.loads(data[0])
self.cache[name] = info
return info
def __setitem__(self, name, data):
with self.cursor() as cursor:
cursor.execute(
"INSERT OR REPLACE INTO profiles (name, data) "
"VALUES (?, ?)", (name, json.dumps(data)))
self.cache[name] = data
def __delitem__(self, name):
with self.cursor() as cursor:
cursor.execute(
"DELETE FROM profiles"
" WHERE name = ?", (name,))
try:
del self.cache[name]
except KeyError:
pass
@classmethod
def create_database(cls, dbpath):
# Initialise the database file with restrictive permissions.
os.close(os.open(dbpath, os.O_CREAT | os.O_APPEND, 0o600))
@classmethod
@contextmanager
def open(cls, dbpath=expanduser("~/.maascli.db")):
"""Load a profiles database.
Called without arguments this will open (and create) a database in the
user's home directory.
**Note** that this returns a context manager which will close the
database on exit, saving if the exit is clean.
"""
# As the effective UID and GID of the user invoking `sudo` (if any)...
try:
with utils.sudo_gid(), utils.sudo_uid():
cls.create_database(dbpath)
except PermissionError:
# Creating the database might fail if $HOME is set to the current
# effective UID's $HOME, but we have permission to change the UID
# to one without permission to access $HOME. So try again without
# changing the GID/UID.
cls.create_database(dbpath)
database = sqlite3.connect(dbpath)
try:
yield cls(database)
except:
raise
else:
database.commit()
finally:
database.close()
|