/usr/lib/python3/dist-packages/curtin/config.py is in python3-curtin 18.1-5-g572ae5d6-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 126 127 128 129 | # This file is part of curtin. See LICENSE file for copyright and license info.
import yaml
import json
ARCHIVE_HEADER = "#curtin-config-archive"
ARCHIVE_TYPE = "text/curtin-config-archive"
CONFIG_HEADER = "#curtin-config"
CONFIG_TYPE = "text/curtin-config"
try:
# python2
_STRING_TYPES = (str, basestring, unicode)
except NameError:
# python3
_STRING_TYPES = (str,)
def merge_config_fp(cfgin, fp):
merge_config_str(cfgin, fp.read())
def merge_config_str(cfgin, cfgstr):
cfg2 = yaml.safe_load(cfgstr)
if not isinstance(cfg2, dict):
raise TypeError("Failed reading config. not a dictionary: %s" % cfgstr)
merge_config(cfgin, cfg2)
def merge_config(cfg, cfg2):
# update cfg by merging cfg2 over the top
for k, v in cfg2.items():
if isinstance(v, dict) and isinstance(cfg.get(k, None), dict):
merge_config(cfg[k], v)
else:
cfg[k] = v
def merge_cmdarg(cfg, cmdarg, delim="/"):
merge_config(cfg, cmdarg2cfg(cmdarg, delim))
def cmdarg2cfg(cmdarg, delim="/"):
if '=' not in cmdarg:
raise ValueError('no "=" in "%s"' % cmdarg)
key, val = cmdarg.split("=", 1)
cfg = {}
cur = cfg
is_json = False
if key.startswith("json:"):
is_json = True
key = key[5:]
items = key.split(delim)
for item in items[:-1]:
cur[item] = {}
cur = cur[item]
if is_json:
try:
val = json.loads(val)
except (ValueError, TypeError):
raise ValueError("setting of key '%s' had invalid json: %s" %
(key, val))
# this would occur if 'json:={"topkey": "topval"}'
if items[-1] == "":
cfg = val
else:
cur[items[-1]] = val
return cfg
def load_config_archive(content):
archive = yaml.load(content)
config = {}
for part in archive:
if isinstance(part, (str,)):
if part.startswith(ARCHIVE_HEADER):
merge_config(config, load_config_archive(part))
elif part.startswith(CONFIG_HEADER):
merge_config_str(config, part)
elif isinstance(part, dict) and isinstance(part.get('content'), str):
payload = part.get('content')
if (part.get('type') == ARCHIVE_TYPE or
payload.startswith(ARCHIVE_HEADER)):
merge_config(config, load_config_archive(payload))
elif (part.get('type') == CONFIG_TYPE or
payload.startswith(CONFIG_HEADER)):
merge_config_str(config, payload)
return config
def load_config(cfg_file):
with open(cfg_file, "r") as fp:
content = fp.read()
if not content.startswith(ARCHIVE_HEADER):
return yaml.safe_load(content)
else:
return load_config_archive(content)
def load_command_config(args, state):
if hasattr(args, 'config') and args.config:
return args.config
else:
# state 'config' points to a file with fully rendered config
cfg_file = state.get('config')
if not cfg_file:
cfg = {}
else:
cfg = load_config(cfg_file)
return cfg
def dump_config(config):
return yaml.dump(config, default_flow_style=False, indent=2)
def value_as_boolean(value):
false_values = (False, None, 0, '0', 'False', 'false', 'None', 'none', '')
return value not in false_values
# vi: ts=4 expandtab syntax=python
|