summaryrefslogtreecommitdiff
path: root/cloudinit/config/cc_apt_configure.py
diff options
context:
space:
mode:
authorJames Falcon <james.falcon@canonical.com>2021-12-15 20:16:38 -0600
committerGitHub <noreply@github.com>2021-12-15 19:16:38 -0700
commitbae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf (patch)
tree1fbb3269fc87e39832e3286ef42eefd2b23fcd44 /cloudinit/config/cc_apt_configure.py
parent2bcf4fa972fde686c2e3141c58e640640b44dd00 (diff)
downloadvyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.tar.gz
vyos-cloud-init-bae9b11da9ed7dd0b16fe5adeaf4774b7cc628cf.zip
Adopt Black and isort (SC-700) (#1157)
Applied Black and isort, fixed any linting issues, updated tox.ini and CI.
Diffstat (limited to 'cloudinit/config/cc_apt_configure.py')
-rw-r--r--cloudinit/config/cc_apt_configure.py618
1 files changed, 340 insertions, 278 deletions
diff --git a/cloudinit/config/cc_apt_configure.py b/cloudinit/config/cc_apt_configure.py
index 2e844c2c..b0728517 100644
--- a/cloudinit/config/cc_apt_configure.py
+++ b/cloudinit/config/cc_apt_configure.py
@@ -10,16 +10,14 @@
import glob
import os
-import re
import pathlib
+import re
from textwrap import dedent
-from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
from cloudinit import gpg
from cloudinit import log as logging
-from cloudinit import subp
-from cloudinit import templater
-from cloudinit import util
+from cloudinit import subp, templater, util
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
from cloudinit.settings import PER_INSTANCE
LOG = logging.getLogger(__name__)
@@ -27,59 +25,46 @@ LOG = logging.getLogger(__name__)
# this will match 'XXX:YYY' (ie, 'cloud-archive:foo' or 'ppa:bar')
ADD_APT_REPO_MATCH = r"^[\w-]+:\w"
-APT_LOCAL_KEYS = '/etc/apt/trusted.gpg'
-APT_TRUSTED_GPG_DIR = '/etc/apt/trusted.gpg.d/'
-CLOUD_INIT_GPG_DIR = '/etc/apt/cloud-init.gpg.d/'
+APT_LOCAL_KEYS = "/etc/apt/trusted.gpg"
+APT_TRUSTED_GPG_DIR = "/etc/apt/trusted.gpg.d/"
+CLOUD_INIT_GPG_DIR = "/etc/apt/cloud-init.gpg.d/"
frequency = PER_INSTANCE
distros = ["ubuntu", "debian"]
mirror_property = {
- 'type': 'array',
- 'items': {
- 'type': 'object',
- 'additionalProperties': False,
- 'required': ['arches'],
- 'properties': {
- 'arches': {
- 'type': 'array',
- 'items': {
- 'type': 'string'
- },
- 'minItems': 1
- },
- 'uri': {
- 'type': 'string',
- 'format': 'uri'
- },
- 'search': {
- 'type': 'array',
- 'items': {
- 'type': 'string',
- 'format': 'uri'
- },
- 'minItems': 1
- },
- 'search_dns': {
- 'type': 'boolean',
+ "type": "array",
+ "items": {
+ "type": "object",
+ "additionalProperties": False,
+ "required": ["arches"],
+ "properties": {
+ "arches": {
+ "type": "array",
+ "items": {"type": "string"},
+ "minItems": 1,
},
- 'keyid': {
- 'type': 'string'
+ "uri": {"type": "string", "format": "uri"},
+ "search": {
+ "type": "array",
+ "items": {"type": "string", "format": "uri"},
+ "minItems": 1,
},
- 'key': {
- 'type': 'string'
+ "search_dns": {
+ "type": "boolean",
},
- 'keyserver': {
- 'type': 'string'
- }
- }
- }
+ "keyid": {"type": "string"},
+ "key": {"type": "string"},
+ "keyserver": {"type": "string"},
+ },
+ },
}
meta = {
- 'id': 'cc_apt_configure',
- 'name': 'Apt Configure',
- 'title': 'Configure apt for the user',
- 'description': dedent("""\
+ "id": "cc_apt_configure",
+ "name": "Apt Configure",
+ "title": "Configure apt for the user",
+ "description": dedent(
+ """\
This module handles both configuration of apt options and adding
source lists. There are configuration options such as
``apt_get_wrapper`` and ``apt_get_command`` that control how
@@ -94,9 +79,12 @@ meta = {
.. note::
For more information about apt configuration, see the
- ``Additional apt configuration`` example."""),
- 'distros': distros,
- 'examples': [dedent("""\
+ ``Additional apt configuration`` example."""
+ ),
+ "distros": distros,
+ "examples": [
+ dedent(
+ """\
apt:
preserve_sources_list: false
disable_suites:
@@ -153,21 +141,24 @@ meta = {
key: |
------BEGIN PGP PUBLIC KEY BLOCK-------
<key data>
- ------END PGP PUBLIC KEY BLOCK-------""")],
- 'frequency': frequency,
+ ------END PGP PUBLIC KEY BLOCK-------"""
+ )
+ ],
+ "frequency": frequency,
}
schema = {
- 'type': 'object',
- 'properties': {
- 'apt': {
- 'type': 'object',
- 'additionalProperties': False,
- 'properties': {
- 'preserve_sources_list': {
- 'type': 'boolean',
- 'default': False,
- 'description': dedent("""\
+ "type": "object",
+ "properties": {
+ "apt": {
+ "type": "object",
+ "additionalProperties": False,
+ "properties": {
+ "preserve_sources_list": {
+ "type": "boolean",
+ "default": False,
+ "description": dedent(
+ """\
By default, cloud-init will generate a new sources
list in ``/etc/apt/sources.list.d`` based on any
changes specified in cloud config. To disable this
@@ -179,15 +170,15 @@ schema = {
all other config keys that would alter
``sources.list`` or ``sources.list.d``,
**except** for additional sources to be added
- to ``sources.list.d``.""")
+ to ``sources.list.d``."""
+ ),
},
- 'disable_suites': {
- 'type': 'array',
- 'items': {
- 'type': 'string'
- },
- 'uniqueItems': True,
- 'description': dedent("""\
+ "disable_suites": {
+ "type": "array",
+ "items": {"type": "string"},
+ "uniqueItems": True,
+ "description": dedent(
+ """\
Entries in the sources list can be disabled using
``disable_suites``, which takes a list of suites
to be disabled. If the string ``$RELEASE`` is
@@ -206,11 +197,13 @@ schema = {
When a suite is disabled using ``disable_suites``,
its entry in ``sources.list`` is not deleted; it
- is just commented out.""")
+ is just commented out."""
+ ),
},
- 'primary': {
+ "primary": {
**mirror_property,
- 'description': dedent("""\
+ "description": dedent(
+ """\
The primary and security archive mirrors can
be specified using the ``primary`` and
``security`` keys, respectively. Both the
@@ -264,27 +257,35 @@ schema = {
``http://archive.ubuntu.com/ubuntu``.
- ``security`` => \
``http://security.ubuntu.com/ubuntu``
- """)
+ """
+ ),
},
- 'security': {
+ "security": {
**mirror_property,
- 'description': dedent("""\
- Please refer to the primary config documentation""")
+ "description": dedent(
+ """\
+ Please refer to the primary config documentation"""
+ ),
},
- 'add_apt_repo_match': {
- 'type': 'string',
- 'default': ADD_APT_REPO_MATCH,
- 'description': dedent("""\
+ "add_apt_repo_match": {
+ "type": "string",
+ "default": ADD_APT_REPO_MATCH,
+ "description": dedent(
+ """\
All source entries in ``apt-sources`` that match
regex in ``add_apt_repo_match`` will be added to
the system using ``add-apt-repository``. If
``add_apt_repo_match`` is not specified, it
- defaults to ``{}``""".format(ADD_APT_REPO_MATCH))
+ defaults to ``{}``""".format(
+ ADD_APT_REPO_MATCH
+ )
+ ),
},
- 'debconf_selections': {
- 'type': 'object',
- 'items': {'type': 'string'},
- 'description': dedent("""\
+ "debconf_selections": {
+ "type": "object",
+ "items": {"type": "string"},
+ "description": dedent(
+ """\
Debconf additional configurations can be specified as a
dictionary under the ``debconf_selections`` config
key, with each key in the dict representing a
@@ -308,11 +309,13 @@ schema = {
For example: \
``ippackage ippackage/ip string 127.0.01``
- """)
+ """
+ ),
},
- 'sources_list': {
- 'type': 'string',
- 'description': dedent("""\
+ "sources_list": {
+ "type": "string",
+ "description": dedent(
+ """\
Specifies a custom template for rendering
``sources.list`` . If no ``sources_list`` template
is given, cloud-init will use sane default. Within
@@ -323,45 +326,55 @@ schema = {
- ``$RELEASE``
- ``$PRIMARY``
- ``$SECURITY``
- - ``$KEY_FILE``""")
+ - ``$KEY_FILE``"""
+ ),
},
- 'conf': {
- 'type': 'string',
- 'description': dedent("""\
+ "conf": {
+ "type": "string",
+ "description": dedent(
+ """\
Specify configuration for apt, such as proxy
configuration. This configuration is specified as a
string. For multiline apt configuration, make sure
- to follow yaml syntax.""")
+ to follow yaml syntax."""
+ ),
},
- 'https_proxy': {
- 'type': 'string',
- 'description': dedent("""\
+ "https_proxy": {
+ "type": "string",
+ "description": dedent(
+ """\
More convenient way to specify https apt proxy.
https proxy url is specified in the format
- ``https://[[user][:pass]@]host[:port]/``.""")
+ ``https://[[user][:pass]@]host[:port]/``."""
+ ),
},
- 'http_proxy': {
- 'type': 'string',
- 'description': dedent("""\
+ "http_proxy": {
+ "type": "string",
+ "description": dedent(
+ """\
More convenient way to specify http apt proxy.
http proxy url is specified in the format
- ``http://[[user][:pass]@]host[:port]/``.""")
+ ``http://[[user][:pass]@]host[:port]/``."""
+ ),
},
- 'proxy': {
- 'type': 'string',
- 'description': 'Alias for defining a http apt proxy.'
+ "proxy": {
+ "type": "string",
+ "description": "Alias for defining a http apt proxy.",
},
- 'ftp_proxy': {
- 'type': 'string',
- 'description': dedent("""\
+ "ftp_proxy": {
+ "type": "string",
+ "description": dedent(
+ """\
More convenient way to specify ftp apt proxy.
ftp proxy url is specified in the format
- ``ftp://[[user][:pass]@]host[:port]/``.""")
+ ``ftp://[[user][:pass]@]host[:port]/``."""
+ ),
},
- 'sources': {
- 'type': 'object',
- 'items': {'type': 'string'},
- 'description': dedent("""\
+ "sources": {
+ "type": "object",
+ "items": {"type": "string"},
+ "description": dedent(
+ """\
Source list entries can be specified as a
dictionary under the ``sources`` config key, with
each key in the dict representing a different source
@@ -394,11 +407,12 @@ schema = {
- ``$PRIMARY``
- ``$SECURITY``
- ``$RELEASE``
- - ``$KEY_FILE``""")
- }
- }
+ - ``$KEY_FILE``"""
+ ),
+ },
+ },
}
- }
+ },
}
__doc__ = get_meta_doc(meta, schema)
@@ -415,18 +429,22 @@ APT_PROXY_FN = "/etc/apt/apt.conf.d/90cloud-init-aptproxy"
DEFAULT_KEYSERVER = "keyserver.ubuntu.com"
# Default archive mirrors
-PRIMARY_ARCH_MIRRORS = {"PRIMARY": "http://archive.ubuntu.com/ubuntu/",
- "SECURITY": "http://security.ubuntu.com/ubuntu/"}
-PORTS_MIRRORS = {"PRIMARY": "http://ports.ubuntu.com/ubuntu-ports",
- "SECURITY": "http://ports.ubuntu.com/ubuntu-ports"}
-PRIMARY_ARCHES = ['amd64', 'i386']
-PORTS_ARCHES = ['s390x', 'arm64', 'armhf', 'powerpc', 'ppc64el', 'riscv64']
+PRIMARY_ARCH_MIRRORS = {
+ "PRIMARY": "http://archive.ubuntu.com/ubuntu/",
+ "SECURITY": "http://security.ubuntu.com/ubuntu/",
+}
+PORTS_MIRRORS = {
+ "PRIMARY": "http://ports.ubuntu.com/ubuntu-ports",
+ "SECURITY": "http://ports.ubuntu.com/ubuntu-ports",
+}
+PRIMARY_ARCHES = ["amd64", "i386"]
+PORTS_ARCHES = ["s390x", "arm64", "armhf", "powerpc", "ppc64el", "riscv64"]
def get_default_mirrors(arch=None, target=None):
"""returns the default mirrors for the target. These depend on the
- architecture, for more see:
- https://wiki.ubuntu.com/UbuntuDevelopment/PackageArchive#Ports"""
+ architecture, for more see:
+ https://wiki.ubuntu.com/UbuntuDevelopment/PackageArchive#Ports"""
if arch is None:
arch = util.get_dpkg_architecture(target)
if arch in PRIMARY_ARCHES:
@@ -438,8 +456,8 @@ def get_default_mirrors(arch=None, target=None):
def handle(name, ocfg, cloud, log, _):
"""process the config for apt_config. This can be called from
- curthooks if a global apt config was provided or via the "apt"
- standalone command."""
+ curthooks if a global apt config was provided or via the "apt"
+ standalone command."""
# keeping code close to curtin codebase via entry handler
target = None
if log is not None:
@@ -447,12 +465,14 @@ def handle(name, ocfg, cloud, log, _):
LOG = log
# feed back converted config, but only work on the subset under 'apt'
ocfg = convert_to_v3_apt_format(ocfg)
- cfg = ocfg.get('apt', {})
+ cfg = ocfg.get("apt", {})
if not isinstance(cfg, dict):
raise ValueError(
"Expected dictionary for 'apt' config, found {config_type}".format(
- config_type=type(cfg)))
+ config_type=type(cfg)
+ )
+ )
validate_cloudconfig_schema(cfg, schema)
apply_debconf_selections(cfg, target)
@@ -463,7 +483,7 @@ def _should_configure_on_empty_apt():
# if no config was provided, should apt configuration be done?
if util.system_is_snappy():
return False, "system is snappy."
- if not (subp.which('apt-get') or subp.which('apt')):
+ if not (subp.which("apt-get") or subp.which("apt")):
return False, "no apt commands."
return True, "Apt is available."
@@ -478,12 +498,12 @@ def apply_apt(cfg, cloud, target):
LOG.debug("handling apt config: %s", cfg)
- release = util.lsb_release(target=target)['codename']
+ release = util.lsb_release(target=target)["codename"]
arch = util.get_dpkg_architecture(target)
mirrors = find_apt_mirror_info(cfg, cloud, arch=arch)
LOG.debug("Apt Mirror info: %s", mirrors)
- if util.is_false(cfg.get('preserve_sources_list', False)):
+ if util.is_false(cfg.get("preserve_sources_list", False)):
add_mirror_keys(cfg, target)
generate_sources_list(cfg, release, mirrors, cloud)
rename_apt_lists(mirrors, target, arch)
@@ -494,25 +514,34 @@ def apply_apt(cfg, cloud, target):
LOG.exception("Failed to apply proxy or apt config info:")
# Process 'apt_source -> sources {dict}'
- if 'sources' in cfg:
+ if "sources" in cfg:
params = mirrors
- params['RELEASE'] = release
- params['MIRROR'] = mirrors["MIRROR"]
+ params["RELEASE"] = release
+ params["MIRROR"] = mirrors["MIRROR"]
matcher = None
- matchcfg = cfg.get('add_apt_repo_match', ADD_APT_REPO_MATCH)
+ matchcfg = cfg.get("add_apt_repo_match", ADD_APT_REPO_MATCH)
if matchcfg:
matcher = re.compile(matchcfg).search
- add_apt_sources(cfg['sources'], cloud, target=target,
- template_params=params, aa_repo_match=matcher)
+ add_apt_sources(
+ cfg["sources"],
+ cloud,
+ target=target,
+ template_params=params,
+ aa_repo_match=matcher,
+ )
def debconf_set_selections(selections, target=None):
- if not selections.endswith(b'\n'):
- selections += b'\n'
- subp.subp(['debconf-set-selections'], data=selections, target=target,
- capture=True)
+ if not selections.endswith(b"\n"):
+ selections += b"\n"
+ subp.subp(
+ ["debconf-set-selections"],
+ data=selections,
+ target=target,
+ capture=True,
+ )
def dpkg_reconfigure(packages, target=None):
@@ -532,12 +561,20 @@ def dpkg_reconfigure(packages, target=None):
unhandled.append(pkg)
if len(unhandled):
- LOG.warning("The following packages were installed and preseeded, "
- "but cannot be unconfigured: %s", unhandled)
+ LOG.warning(
+ "The following packages were installed and preseeded, "
+ "but cannot be unconfigured: %s",
+ unhandled,
+ )
if len(to_config):
- subp.subp(['dpkg-reconfigure', '--frontend=noninteractive'] +
- list(to_config), data=None, target=target, capture=True)
+ subp.subp(
+ ["dpkg-reconfigure", "--frontend=noninteractive"]
+ + list(to_config),
+ data=None,
+ target=target,
+ capture=True,
+ )
def apply_debconf_selections(cfg, target=None):
@@ -546,13 +583,12 @@ def apply_debconf_selections(cfg, target=None):
# set1: |
# cloud-init cloud-init/datasources multiselect MAAS
# set2: pkg pkg/value string bar
- selsets = cfg.get('debconf_selections')
+ selsets = cfg.get("debconf_selections")
if not selsets:
LOG.debug("debconf_selections was not set in config")
return
- selections = '\n'.join(
- [selsets[key] for key in sorted(selsets.keys())])
+ selections = "\n".join([selsets[key] for key in sorted(selsets.keys())])
debconf_set_selections(selections.encode(), target=target)
# get a complete list of packages listed in input
@@ -579,7 +615,8 @@ def apply_debconf_selections(cfg, target=None):
def clean_cloud_init(target):
"""clean out any local cloud-init config"""
flist = glob.glob(
- subp.target_path(target, "/etc/cloud/cloud.cfg.d/*dpkg*"))
+ subp.target_path(target, "/etc/cloud/cloud.cfg.d/*dpkg*")
+ )
LOG.debug("cleaning cloud-init config from: %s", flist)
for dpkg_cfg in flist:
@@ -588,18 +625,18 @@ def clean_cloud_init(target):
def mirrorurl_to_apt_fileprefix(mirror):
"""mirrorurl_to_apt_fileprefix
- Convert a mirror url to the file prefix used by apt on disk to
- store cache information for that mirror.
- To do so do:
- - take off ???://
- - drop tailing /
- - convert in string / to _"""
+ Convert a mirror url to the file prefix used by apt on disk to
+ store cache information for that mirror.
+ To do so do:
+ - take off ???://
+ - drop tailing /
+ - convert in string / to _"""
string = mirror
if string.endswith("/"):
string = string[0:-1]
pos = string.find("://")
if pos >= 0:
- string = string[pos + 3:]
+ string = string[pos + 3 :]
string = string.replace("/", "_")
return string
@@ -631,8 +668,8 @@ def rename_apt_lists(new_mirrors, target, arch):
def mirror_to_placeholder(tmpl, mirror, placeholder):
"""mirror_to_placeholder
- replace the specified mirror in a template with a placeholder string
- Checks for existance of the expected mirror and warns if not found"""
+ replace the specified mirror in a template with a placeholder string
+ Checks for existance of the expected mirror and warns if not found"""
if mirror not in tmpl:
LOG.warning("Expected mirror '%s' not found in: %s", mirror, tmpl)
return tmpl.replace(mirror, placeholder)
@@ -640,13 +677,15 @@ def mirror_to_placeholder(tmpl, mirror, placeholder):
def map_known_suites(suite):
"""there are a few default names which will be auto-extended.
- This comes at the inability to use those names literally as suites,
- but on the other hand increases readability of the cfg quite a lot"""
- mapping = {'updates': '$RELEASE-updates',
- 'backports': '$RELEASE-backports',
- 'security': '$RELEASE-security',
- 'proposed': '$RELEASE-proposed',
- 'release': '$RELEASE'}
+ This comes at the inability to use those names literally as suites,
+ but on the other hand increases readability of the cfg quite a lot"""
+ mapping = {
+ "updates": "$RELEASE-updates",
+ "backports": "$RELEASE-backports",
+ "security": "$RELEASE-security",
+ "proposed": "$RELEASE-proposed",
+ "release": "$RELEASE",
+ }
try:
retsuite = mapping[suite]
except KeyError:
@@ -656,14 +695,14 @@ def map_known_suites(suite):
def disable_suites(disabled, src, release):
"""reads the config for suites to be disabled and removes those
- from the template"""
+ from the template"""
if not disabled:
return src
retsrc = src
for suite in disabled:
suite = map_known_suites(suite)
- releasesuite = templater.render_string(suite, {'RELEASE': release})
+ releasesuite = templater.render_string(suite, {"RELEASE": release})
LOG.debug("Disabling suite %s as %s", suite, releasesuite)
newsrc = ""
@@ -685,7 +724,7 @@ def disable_suites(disabled, src, release):
break
if cols[pcol] == releasesuite:
- line = '# suite disabled by cloud-init: %s' % line
+ line = "# suite disabled by cloud-init: %s" % line
newsrc += line
retsrc = newsrc
@@ -694,36 +733,38 @@ def disable_suites(disabled, src, release):
def add_mirror_keys(cfg, target):
"""Adds any keys included in the primary/security mirror clauses"""
- for key in ('primary', 'security'):
+ for key in ("primary", "security"):
for mirror in cfg.get(key, []):
add_apt_key(mirror, target, file_name=key)
def generate_sources_list(cfg, release, mirrors, cloud):
"""generate_sources_list
- create a source.list file based on a custom or default template
- by replacing mirrors and release in the template"""
+ create a source.list file based on a custom or default template
+ by replacing mirrors and release in the template"""
aptsrc = "/etc/apt/sources.list"
- params = {'RELEASE': release, 'codename': release}
+ params = {"RELEASE": release, "codename": release}
for k in mirrors:
params[k] = mirrors[k]
params[k.lower()] = mirrors[k]
- tmpl = cfg.get('sources_list', None)
+ tmpl = cfg.get("sources_list", None)
if tmpl is None:
LOG.info("No custom template provided, fall back to builtin")
- template_fn = cloud.get_template_filename('sources.list.%s' %
- (cloud.distro.name))
+ template_fn = cloud.get_template_filename(
+ "sources.list.%s" % (cloud.distro.name)
+ )
if not template_fn:
- template_fn = cloud.get_template_filename('sources.list')
+ template_fn = cloud.get_template_filename("sources.list")
if not template_fn:
- LOG.warning("No template found, "
- "not rendering /etc/apt/sources.list")
+ LOG.warning(
+ "No template found, not rendering /etc/apt/sources.list"
+ )
return
tmpl = util.load_file(template_fn)
rendered = templater.render_string(tmpl, params)
- disabled = disable_suites(cfg.get('disable_suites'), rendered, release)
+ disabled = disable_suites(cfg.get("disable_suites"), rendered, release)
util.write_file(aptsrc, disabled, mode=0o644)
@@ -735,7 +776,7 @@ def add_apt_key_raw(key, file_name, hardened=False, target=None):
LOG.debug("Adding key:\n'%s'", key)
try:
name = pathlib.Path(file_name).stem
- return apt_key('add', output_file=name, data=key, hardened=hardened)
+ return apt_key("add", output_file=name, data=key, hardened=hardened)
except subp.ProcessExecutionError:
LOG.exception("failed to add apt GPG Key to apt keyring")
raise
@@ -747,26 +788,26 @@ def add_apt_key(ent, target=None, hardened=False, file_name=None):
Supports raw keys or keyid's
The latter will as a first step fetched to get the raw key
"""
- if 'keyid' in ent and 'key' not in ent:
+ if "keyid" in ent and "key" not in ent:
keyserver = DEFAULT_KEYSERVER
- if 'keyserver' in ent:
- keyserver = ent['keyserver']
+ if "keyserver" in ent:
+ keyserver = ent["keyserver"]
- ent['key'] = gpg.getkeybyid(ent['keyid'], keyserver)
+ ent["key"] = gpg.getkeybyid(ent["keyid"], keyserver)
- if 'key' in ent:
+ if "key" in ent:
return add_apt_key_raw(
- ent['key'],
- file_name or ent['filename'],
- hardened=hardened)
+ ent["key"], file_name or ent["filename"], hardened=hardened
+ )
def update_packages(cloud):
cloud.distro.update_package_sources()
-def add_apt_sources(srcdict, cloud, target=None, template_params=None,
- aa_repo_match=None):
+def add_apt_sources(
+ srcdict, cloud, target=None, template_params=None, aa_repo_match=None
+):
"""
install keys and repo source .list files defined in 'sources'
@@ -795,33 +836,34 @@ def add_apt_sources(srcdict, cloud, target=None, template_params=None,
template_params = {}
if aa_repo_match is None:
- raise ValueError('did not get a valid repo matcher')
+ raise ValueError("did not get a valid repo matcher")
if not isinstance(srcdict, dict):
- raise TypeError('unknown apt format: %s' % (srcdict))
+ raise TypeError("unknown apt format: %s" % (srcdict))
for filename in srcdict:
ent = srcdict[filename]
LOG.debug("adding source/key '%s'", ent)
- if 'filename' not in ent:
- ent['filename'] = filename
+ if "filename" not in ent:
+ ent["filename"] = filename
- if 'source' in ent and '$KEY_FILE' in ent['source']:
+ if "source" in ent and "$KEY_FILE" in ent["source"]:
key_file = add_apt_key(ent, target, hardened=True)
- template_params['KEY_FILE'] = key_file
+ template_params["KEY_FILE"] = key_file
else:
key_file = add_apt_key(ent, target)
- if 'source' not in ent:
+ if "source" not in ent:
continue
- source = ent['source']
+ source = ent["source"]
source = templater.render_string(source, template_params)
- if not ent['filename'].startswith("/"):
- ent['filename'] = os.path.join("/etc/apt/sources.list.d/",
- ent['filename'])
- if not ent['filename'].endswith(".list"):
- ent['filename'] += ".list"
+ if not ent["filename"].startswith("/"):
+ ent["filename"] = os.path.join(
+ "/etc/apt/sources.list.d/", ent["filename"]
+ )
+ if not ent["filename"].endswith(".list"):
+ ent["filename"] += ".list"
if aa_repo_match(source):
try:
@@ -831,7 +873,7 @@ def add_apt_sources(srcdict, cloud, target=None, template_params=None,
raise
continue
- sourcefn = subp.target_path(target, ent['filename'])
+ sourcefn = subp.target_path(target, ent["filename"])
try:
contents = "%s\n" % (source)
util.write_file(sourcefn, contents, omode="a")
@@ -850,14 +892,14 @@ def convert_v1_to_v2_apt_format(srclist):
if isinstance(srclist, list):
LOG.debug("apt config: convert V1 to V2 format (source list to dict)")
for srcent in srclist:
- if 'filename' not in srcent:
+ if "filename" not in srcent:
# file collides for multiple !filename cases for compatibility
# yet we need them all processed, so not same dictionary key
- srcent['filename'] = "cloud_config_sources.list"
+ srcent["filename"] = "cloud_config_sources.list"
key = util.rand_dict_key(srcdict, "cloud_config_sources.list")
else:
# all with filename use that as key (matching new format)
- key = srcent['filename']
+ key = srcent["filename"]
srcdict[key] = srcent
elif isinstance(srclist, dict):
srcdict = srclist
@@ -869,7 +911,7 @@ def convert_v1_to_v2_apt_format(srclist):
def convert_key(oldcfg, aptcfg, oldkey, newkey):
"""convert an old key to the new one if the old one exists
- returns true if a key was found and converted"""
+ returns true if a key was found and converted"""
if oldcfg.get(oldkey, None) is not None:
aptcfg[newkey] = oldcfg.get(oldkey)
del oldcfg[oldkey]
@@ -879,33 +921,37 @@ def convert_key(oldcfg, aptcfg, oldkey, newkey):
def convert_mirror(oldcfg, aptcfg):
"""convert old apt_mirror keys into the new more advanced mirror spec"""
- keymap = [('apt_mirror', 'uri'),
- ('apt_mirror_search', 'search'),
- ('apt_mirror_search_dns', 'search_dns')]
+ keymap = [
+ ("apt_mirror", "uri"),
+ ("apt_mirror_search", "search"),
+ ("apt_mirror_search_dns", "search_dns"),
+ ]
converted = False
- newmcfg = {'arches': ['default']}
+ newmcfg = {"arches": ["default"]}
for oldkey, newkey in keymap:
if convert_key(oldcfg, newmcfg, oldkey, newkey):
converted = True
# only insert new style config if anything was converted
if converted:
- aptcfg['primary'] = [newmcfg]
+ aptcfg["primary"] = [newmcfg]
def convert_v2_to_v3_apt_format(oldcfg):
"""convert old to new keys and adapt restructured mirror spec"""
- mapoldkeys = {'apt_sources': 'sources',
- 'apt_mirror': None,
- 'apt_mirror_search': None,
- 'apt_mirror_search_dns': None,
- 'apt_proxy': 'proxy',
- 'apt_http_proxy': 'http_proxy',
- 'apt_ftp_proxy': 'https_proxy',
- 'apt_https_proxy': 'ftp_proxy',
- 'apt_preserve_sources_list': 'preserve_sources_list',
- 'apt_custom_sources_list': 'sources_list',
- 'add_apt_repo_match': 'add_apt_repo_match'}
+ mapoldkeys = {
+ "apt_sources": "sources",
+ "apt_mirror": None,
+ "apt_mirror_search": None,
+ "apt_mirror_search_dns": None,
+ "apt_proxy": "proxy",
+ "apt_http_proxy": "http_proxy",
+ "apt_ftp_proxy": "https_proxy",
+ "apt_https_proxy": "ftp_proxy",
+ "apt_preserve_sources_list": "preserve_sources_list",
+ "apt_custom_sources_list": "sources_list",
+ "add_apt_repo_match": "add_apt_repo_match",
+ }
needtoconvert = []
for oldkey in mapoldkeys:
if oldkey in oldcfg:
@@ -917,11 +963,13 @@ def convert_v2_to_v3_apt_format(oldcfg):
# no old config, so no new one to be created
if not needtoconvert:
return oldcfg
- LOG.debug("apt config: convert V2 to V3 format for keys '%s'",
- ", ".join(needtoconvert))
+ LOG.debug(
+ "apt config: convert V2 to V3 format for keys '%s'",
+ ", ".join(needtoconvert),
+ )
# if old AND new config are provided, prefer the new one (LP #1616831)
- newaptcfg = oldcfg.get('apt', None)
+ newaptcfg = oldcfg.get("apt", None)
if newaptcfg is not None:
LOG.debug("apt config: V1/2 and V3 format specified, preferring V3")
for oldkey in needtoconvert:
@@ -932,10 +980,11 @@ def convert_v2_to_v3_apt_format(oldcfg):
# no simple mapping or no collision on this particular key
continue
if verify != newaptcfg[newkey]:
- raise ValueError("Old and New apt format defined with unequal "
- "values %s vs %s @ %s" % (verify,
- newaptcfg[newkey],
- oldkey))
+ raise ValueError(
+ "Old and New apt format defined with unequal "
+ "values %s vs %s @ %s"
+ % (verify, newaptcfg[newkey], oldkey)
+ )
# return conf after clearing conflicting V1/2 keys
return oldcfg
@@ -955,17 +1004,17 @@ def convert_v2_to_v3_apt_format(oldcfg):
raise ValueError("old apt key '%s' left after conversion" % oldkey)
# insert new format into config and return full cfg with only v3 content
- oldcfg['apt'] = aptcfg
+ oldcfg["apt"] = aptcfg
return oldcfg
def convert_to_v3_apt_format(cfg):
"""convert the old list based format to the new dict based one. After that
- convert the old dict keys/format to v3 a.k.a 'new apt config'"""
+ convert the old dict keys/format to v3 a.k.a 'new apt config'"""
# V1 -> V2, the apt_sources entry from list to dict
- apt_sources = cfg.get('apt_sources', None)
+ apt_sources = cfg.get("apt_sources", None)
if apt_sources is not None:
- cfg['apt_sources'] = convert_v1_to_v2_apt_format(apt_sources)
+ cfg["apt_sources"] = convert_v1_to_v2_apt_format(apt_sources)
# V2 -> V3, move all former globals under the "apt" key
# Restructure into new key names and mirror hierarchy
@@ -997,7 +1046,12 @@ def search_for_mirror_dns(configured, mirrortype, cfg, cloud):
if mydom:
doms.append(".%s" % mydom)
- doms.extend((".localdomain", "",))
+ doms.extend(
+ (
+ ".localdomain",
+ "",
+ )
+ )
mirror_list = []
distro = cloud.distro.name
@@ -1012,12 +1066,11 @@ def search_for_mirror_dns(configured, mirrortype, cfg, cloud):
def update_mirror_info(pmirror, smirror, arch, cloud):
"""sets security mirror to primary if not defined.
- returns defaults if no mirrors are defined"""
+ returns defaults if no mirrors are defined"""
if pmirror is not None:
if smirror is None:
smirror = pmirror
- return {'PRIMARY': pmirror,
- 'SECURITY': smirror}
+ return {"PRIMARY": pmirror, "SECURITY": smirror}
# None specified at all, get default mirrors from cloud
mirror_info = cloud.datasource.get_package_mirror_info()
@@ -1026,8 +1079,8 @@ def update_mirror_info(pmirror, smirror, arch, cloud):
# arbitrary key/value pairs including 'primary' and 'security' keys.
# caller expects dict with PRIMARY and SECURITY.
m = mirror_info.copy()
- m['PRIMARY'] = m['primary']
- m['SECURITY'] = m['security']
+ m["PRIMARY"] = m["primary"]
+ m["SECURITY"] = m["security"]
return m
@@ -1037,7 +1090,7 @@ def update_mirror_info(pmirror, smirror, arch, cloud):
def get_arch_mirrorconfig(cfg, mirrortype, arch):
"""out of a list of potential mirror configurations select
- and return the one matching the architecture (or default)"""
+ and return the one matching the architecture (or default)"""
# select the mirror specification (if-any)
mirror_cfg_list = cfg.get(mirrortype, None)
if mirror_cfg_list is None:
@@ -1056,8 +1109,8 @@ def get_arch_mirrorconfig(cfg, mirrortype, arch):
def get_mirror(cfg, mirrortype, arch, cloud):
"""pass the three potential stages of mirror specification
- returns None is neither of them found anything otherwise the first
- hit is returned"""
+ returns None is neither of them found anything otherwise the first
+ hit is returned"""
mcfg = get_arch_mirrorconfig(cfg, mirrortype, arch)
if mcfg is None:
return None
@@ -1073,18 +1126,19 @@ def get_mirror(cfg, mirrortype, arch, cloud):
# fallback to search_dns if specified
if mirror is None:
# list of mirrors to try to resolve
- mirror = search_for_mirror_dns(mcfg.get("search_dns", None),
- mirrortype, cfg, cloud)
+ mirror = search_for_mirror_dns(
+ mcfg.get("search_dns", None), mirrortype, cfg, cloud
+ )
return mirror
def find_apt_mirror_info(cfg, cloud, arch=None):
"""find_apt_mirror_info
- find an apt_mirror given the cfg provided.
- It can check for separate config of primary and security mirrors
- If only primary is given security is assumed to be equal to primary
- If the generic apt_mirror is given that is defining for both
+ find an apt_mirror given the cfg provided.
+ It can check for separate config of primary and security mirrors
+ If only primary is given security is assumed to be equal to primary
+ If the generic apt_mirror is given that is defining for both
"""
if arch is None:
@@ -1105,32 +1159,35 @@ def find_apt_mirror_info(cfg, cloud, arch=None):
def apply_apt_config(cfg, proxy_fname, config_fname):
"""apply_apt_config
- Applies any apt*proxy config from if specified
+ Applies any apt*proxy config from if specified
"""
# Set up any apt proxy
- cfgs = (('proxy', 'Acquire::http::Proxy "%s";'),
- ('http_proxy', 'Acquire::http::Proxy "%s";'),
- ('ftp_proxy', 'Acquire::ftp::Proxy "%s";'),
- ('https_proxy', 'Acquire::https::Proxy "%s";'))
+ cfgs = (
+ ("proxy", 'Acquire::http::Proxy "%s";'),
+ ("http_proxy", 'Acquire::http::Proxy "%s";'),
+ ("ftp_proxy", 'Acquire::ftp::Proxy "%s";'),
+ ("https_proxy", 'Acquire::https::Proxy "%s";'),
+ )
proxies = [fmt % cfg.get(name) for (name, fmt) in cfgs if cfg.get(name)]
if len(proxies):
LOG.debug("write apt proxy info to %s", proxy_fname)
- util.write_file(proxy_fname, '\n'.join(proxies) + '\n')
+ util.write_file(proxy_fname, "\n".join(proxies) + "\n")
elif os.path.isfile(proxy_fname):
util.del_file(proxy_fname)
LOG.debug("no apt proxy configured, removed %s", proxy_fname)
- if cfg.get('conf', None):
+ if cfg.get("conf", None):
LOG.debug("write apt config info to %s", config_fname)
- util.write_file(config_fname, cfg.get('conf'))
+ util.write_file(config_fname, cfg.get("conf"))
elif os.path.isfile(config_fname):
util.del_file(config_fname)
LOG.debug("no apt config configured, removed %s", config_fname)
-def apt_key(command, output_file=None, data=None, hardened=False,
- human_output=True):
+def apt_key(
+ command, output_file=None, data=None, hardened=False, human_output=True
+):
"""apt-key replacement
commands implemented: 'add', 'list', 'finger'
@@ -1153,32 +1210,36 @@ def apt_key(command, output_file=None, data=None, hardened=False,
key_files = [APT_LOCAL_KEYS] if os.path.isfile(APT_LOCAL_KEYS) else []
for file in os.listdir(APT_TRUSTED_GPG_DIR):
- if file.endswith('.gpg') or file.endswith('.asc'):
+ if file.endswith(".gpg") or file.endswith(".asc"):
key_files.append(APT_TRUSTED_GPG_DIR + file)
- return key_files if key_files else ''
+ return key_files if key_files else ""
def apt_key_add():
"""apt-key add <file>
returns filepath to new keyring, or '/dev/null' when an error occurs
"""
- file_name = '/dev/null'
+ file_name = "/dev/null"
if not output_file:
util.logexc(
- LOG, 'Unknown filename, failed to add key: "{}"'.format(data))
+ LOG, 'Unknown filename, failed to add key: "{}"'.format(data)
+ )
else:
try:
- key_dir = \
+ key_dir = (
CLOUD_INIT_GPG_DIR if hardened else APT_TRUSTED_GPG_DIR
+ )
stdout = gpg.dearmor(data)
- file_name = '{}{}.gpg'.format(key_dir, output_file)
+ file_name = "{}{}.gpg".format(key_dir, output_file)
util.write_file(file_name, stdout)
except subp.ProcessExecutionError:
- util.logexc(LOG, 'Gpg error, failed to add key: {}'.format(
- data))
+ util.logexc(
+ LOG, "Gpg error, failed to add key: {}".format(data)
+ )
except UnicodeDecodeError:
- util.logexc(LOG, 'Decode error, failed to add key: {}'.format(
- data))
+ util.logexc(
+ LOG, "Decode error, failed to add key: {}".format(data)
+ )
return file_name
def apt_key_list():
@@ -1193,19 +1254,20 @@ def apt_key(command, output_file=None, data=None, hardened=False,
key_list.append(gpg.list(key_file, human_output=human_output))
except subp.ProcessExecutionError as error:
LOG.warning('Failed to list key "%s": %s', key_file, error)
- return '\n'.join(key_list)
+ return "\n".join(key_list)
- if command == 'add':
+ if command == "add":
return apt_key_add()
- elif command == 'finger' or command == 'list':
+ elif command == "finger" or command == "list":
return apt_key_list()
else:
raise ValueError(
- 'apt_key() commands add, list, and finger are currently supported')
+ "apt_key() commands add, list, and finger are currently supported"
+ )
CONFIG_CLEANERS = {
- 'cloud-init': clean_cloud_init,
+ "cloud-init": clean_cloud_init,
}
# vi: ts=4 expandtab