/usr/share/pyshared/provisioningserver/config.py is in python-maas-provisioningserver 1.2+bzr1373+dfsg-0ubuntu1~12.04.6.
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 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | # Copyright 2012 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""MAAS Provisioning Configuration."""
from __future__ import (
absolute_import,
print_function,
unicode_literals,
)
__metaclass__ = type
__all__ = [
"Config",
]
from getpass import getuser
from os import environ
import os.path
from threading import RLock
from formencode import Schema
from formencode.declarative import DeclarativeMeta
from formencode.validators import (
Int,
RequireIfPresent,
String,
)
import yaml
class ConfigOops(Schema):
"""Configuration validator for OOPS options."""
if_key_missing = None
directory = String(if_missing=b"")
reporter = String(if_missing=b"")
chained_validators = (
RequireIfPresent("reporter", present="directory"),
)
class ConfigBroker(Schema):
"""Configuration validator for message broker options."""
if_key_missing = None
host = String(if_missing=b"localhost")
port = Int(min=1, max=65535, if_missing=5673)
username = String(if_missing=getuser())
password = String(if_missing=b"test")
vhost = String(if_missing="/")
class ConfigTFTP(Schema):
"""Configuration validator for the TFTP service."""
if_key_missing = None
root = String(if_missing="/var/lib/maas/tftp")
port = Int(min=1, max=65535, if_missing=69)
generator = String(if_missing=b"http://localhost/MAAS/api/1.0/pxeconfig/")
class ConfigBootEphemeral(Schema):
"""Configuration validator for ephemeral boot configuration."""
if_key_missing = None
directory = String(if_missing="/var/lib/maas/ephemeral")
class ConfigBoot(Schema):
"""Configuration validator for boot configuration."""
if_key_missing = None
ephemeral = ConfigBootEphemeral
class ConfigMeta(DeclarativeMeta):
"""Metaclass for the root configuration schema."""
def _get_default_filename(cls):
# Get the configuration filename from the environment. Failing that,
# look for the configuration in its default locations.
configured_location = environ.get("MAAS_PROVISIONING_SETTINGS")
if configured_location is not None:
return configured_location
for filename in ['etc/pserv.yaml', '/etc/maas/pserv.yaml']:
if os.path.isfile(filename):
return filename
# Oh well. Just return the last filename we tried, and let the
# caller report that that file was not found.
return filename
def _set_default_filename(cls, filename):
# Set the configuration filename in the environment.
environ["MAAS_PROVISIONING_SETTINGS"] = filename
def _delete_default_filename(cls):
# Remove any setting of the configuration filename from the
# environment.
environ.pop("MAAS_PROVISIONING_SETTINGS", None)
DEFAULT_FILENAME = property(
_get_default_filename, _set_default_filename,
_delete_default_filename, doc=(
"The default config file to load. Refers to "
"MAAS_PROVISIONING_SETTINGS in the environment."))
class Config(Schema):
"""Configuration validator."""
__metaclass__ = ConfigMeta
if_key_missing = None
logfile = String(if_empty=b"pserv.log", if_missing=b"pserv.log")
oops = ConfigOops
broker = ConfigBroker
tftp = ConfigTFTP
boot = ConfigBoot
@classmethod
def parse(cls, stream):
"""Load a YAML configuration from `stream` and validate."""
return cls.to_python(yaml.safe_load(stream))
@classmethod
def load(cls, filename=None):
"""Load a YAML configuration from `filename` and validate."""
if filename is None:
filename = cls.DEFAULT_FILENAME
with open(filename, "rb") as stream:
return cls.parse(stream)
_cache = {}
_cache_lock = RLock()
@classmethod
def load_from_cache(cls, filename=None):
"""Load or return a previously loaded configuration.
This is thread-safe, so is okay to use from Django, for example.
"""
if filename is None:
filename = cls.DEFAULT_FILENAME
filename = os.path.abspath(filename)
with cls._cache_lock:
if filename not in cls._cache:
with open(filename, "rb") as stream:
cls._cache[filename] = cls.parse(stream)
return cls._cache[filename]
@classmethod
def field(target, *steps):
"""Obtain a field by following `steps`."""
for step in steps:
target = target.fields[step]
return target
|