ACIL FM

Nama	Ukuran	Aksi
analyze	-	Open
cmd	-	Open
config	-	Open
distros	-	Open
filters	-	Open
handlers	-	Open
log	-	Open
mergers	-	Open
net	-	Open
reporting	-	Open
sources	-	Open
__pycache__	-	Open
apport.py	8.27 MB	View DL Edit
atomic_helper.py	2.79 MB	View DL Edit
cloud.py	3.71 MB	View DL Edit
dmi.py	7.86 MB	View DL Edit
event.py	2 MB	View DL Edit
features.py	4.87 MB	View DL Edit
gpg.py	7.99 MB	View DL Edit
helpers.py	16.16 MB	View DL Edit
importer.py	2.43 MB	View DL Edit
lifecycle.py	7.78 MB	View DL Edit
netinfo.py	24.02 MB	View DL Edit
performance.py	3.1 MB	View DL Edit
persistence.py	2.52 MB	View DL Edit
registry.py	1022 B	View DL Edit
safeyaml.py	10.11 MB	View DL Edit
settings.py	2.15 MB	View DL Edit
signal_handler.py	1.75 MB	View DL Edit
simpletable.py	1.93 MB	View DL Edit
socket.py	5.93 MB	View DL Edit
ssh_util.py	22.22 MB	View DL Edit
stages.py	41.53 MB	View DL Edit
subp.py	12.36 MB	View DL Edit
templater.py	7.8 MB	View DL Edit
temp_utils.py	2.94 MB	View DL Edit
type_utils.py	703 B	View DL Edit
url_helper.py	34.7 MB	View DL Edit
user_data.py	14.44 MB	View DL Edit
util.py	91.22 MB	View DL Edit
version.py	564 B	View DL Edit
warnings.py	3.76 MB	View DL Edit
__init__.py	0 B	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/cloudinit/gpg.py

# Copyright (C) 2016 Canonical Ltd.
#
# Author: Scott Moser <scott.moser@canonical.com>
# Author: Christian Ehrhardt <christian.ehrhardt@canonical.com>
#
# This file is part of cloud-init. See LICENSE file for license information.

"""gpg.py - Collection of gpg key related functions"""

import logging
import os
import re
import signal
import time
from tempfile import TemporaryDirectory
from typing import Dict, Optional

from cloudinit import subp

LOG = logging.getLogger(__name__)

HOME = "GNUPGHOME"

class GPG:
    def __init__(self):
        self.gpg_started = False
        self._env = {}
        self.temp_dir = TemporaryDirectory()

def __enter__(self):
        return self

@property
    def env(self) -> Dict[str, str]:
        """when this env property gets invoked, set up our temporary
        directory, and also set gpg_started to tell the cleanup()
        method whether or not

why put this here and not in __init__? pytest seems unhappy
        and it's not obvious how to work around it
        """
        if self._env:
            return self._env
        self.gpg_started = True
        self._env = {HOME: self.temp_dir.name}
        return self._env

def __exit__(self, exc_typ, exc_value, traceback):
        self.cleanup()

def cleanup(self) -> None:
        """cleanup the gpg temporary directory and kill gpg"""
        self.kill_gpg()
        if self.temp_dir and os.path.isdir(self.temp_dir.name):
            self.temp_dir.cleanup()

def export_armour(self, key: str) -> Optional[str]:
        """Export gpg key, armoured key gets returned"""
        try:
            return subp.subp(
                ["gpg", "--export", "--armour", key],
                capture=True,
                update_env=self.env,
            ).stdout
        except subp.ProcessExecutionError as error:
            # debug, since it happens for any key not on the system initially
            LOG.debug('Failed to export armoured key "%s": %s', key, error)
        return None

def dearmor(self, key: str) -> str:
        """Dearmor gpg key, dearmored key gets returned

note: man gpg(1) makes no mention of an --armour spelling, only --armor
        """
        return subp.subp(
            ["gpg", "--dearmor"], data=key, decode=False, update_env=self.env
        ).stdout

def list_keys(self, key_file: str, human_output=False) -> str:
        """List keys from a keyring with fingerprints. Default to a
        stable machine parseable format.

@param key_file: a string containing a filepath to a key
        @param human_output: return output intended for human parsing
        """
        cmd = [
            "gpg",
            "--no-options",
            "--with-fingerprint",
            "--no-default-keyring",
            "--list-keys",
            "--keyring",
        ]
        if not human_output:
            cmd.append("--with-colons")

cmd.append(key_file)
        stdout, stderr = subp.subp(cmd, update_env=self.env, capture=True)
        if stderr:
            LOG.warning(
                'Failed to export armoured key "%s": %s', key_file, stderr
            )
        return stdout

def recv_key(self, key: str, keyserver: str, retries=(1, 1)) -> None:
        """Receive gpg key from the specified keyserver.

Retries are done by default because keyservers can be unreliable.
        Additionally, there is no way to determine the difference between
        a non-existent key and a failure.  In both cases gpg (at least 2.2.4)
        exits with status 2 and stderr: "keyserver receive failed: No data"
        It is assumed that a key provided to cloud-init exists on the keyserver
        so re-trying makes better sense than failing.

@param key: a string key fingerprint (as passed to gpg --recv-keys).
        @param keyserver: the keyserver to request keys from.
        @param retries: an iterable of sleep lengths for retries.
        Use None to indicate no retries."""
        LOG.debug("Importing key '%s' from keyserver '%s'", key, keyserver)
        trynum = 0
        error = None
        sleeps = iter(retries or [])
        while True:
            trynum += 1
            try:
                subp.subp(
                    [
                        "gpg",
                        "--no-tty",
                        "--keyserver=%s" % keyserver,
                        "--recv-keys",
                        key,
                    ],
                    capture=True,
                    update_env=self.env,
                )
                LOG.debug(
                    "Imported key '%s' from keyserver '%s' on try %d",
                    key,
                    keyserver,
                    trynum,
                )
                return
            except subp.ProcessExecutionError as e:
                error = e
            try:
                naplen = next(sleeps)
                LOG.debug(
                    "Import failed with exit code %d, will try again in %ss",
                    error.exit_code,
                    naplen,
                )
                time.sleep(naplen)
            except StopIteration as e:
                raise ValueError(
                    "Failed to import key '%s' from keyserver '%s' "
                    "after %d tries: %s" % (key, keyserver, trynum, error)
                ) from e

def delete_key(self, key: str) -> None:
        """Delete the specified key from the local gpg ring"""
        try:
            subp.subp(
                ["gpg", "--batch", "--yes", "--delete-keys", key],
                capture=True,
                update_env=self.env,
            )
        except subp.ProcessExecutionError as error:
            LOG.warning('Failed delete key "%s": %s', key, error)

def getkeybyid(
        self, keyid: str, keyserver: str = "keyserver.ubuntu.com"
    ) -> Optional[str]:
        """get gpg keyid from keyserver"""
        armour = self.export_armour(keyid)
        if not armour:
            try:
                self.recv_key(keyid, keyserver=keyserver)
                armour = self.export_armour(keyid)
            except ValueError:
                LOG.exception("Failed to obtain gpg key %s", keyid)
                raise
            finally:
                # delete just imported key to leave environment as it
                # was before
                self.delete_key(keyid)
        return armour

def kill_gpg(self) -> None:
        """killing with gpgconf is best practice, but when it isn't available
        failover is possible

GH: 4344 - stop gpg-agent/dirmgr daemons spawned by gpg
        key imports. Daemons spawned by cloud-config.service on systemd
        v253 report (running)
        """
        try:
            if not self.gpg_started:
                return
            if subp.which("gpgconf"):
                gpg_process_out = subp.subp(
                    ["gpgconf", "--kill", "all"],
                    capture=True,
                    update_env=self.env,
                ).stdout
            else:
                gpg_process_out = subp.subp(
                    [
                        "ps",
                        "-o",
                        "ppid,pid",
                        "-C",
                        "keyboxd",
                        "-C",
                        "dirmngr",
                        "-C",
                        "gpg-agent",
                    ],
                    capture=True,
                    rcs=[0, 1],
                ).stdout
                gpg_pids = re.findall(
                    r"(?P<ppid>\d+)\s+(?P<pid>\d+)", gpg_process_out
                )
                root_gpg_pids = [
                    int(pid[1]) for pid in gpg_pids if pid[0] == "1"
                ]
                if root_gpg_pids:
                    LOG.debug(
                        "Killing gpg-agent and dirmngr pids: %s", root_gpg_pids
                    )
                for gpg_pid in root_gpg_pids:
                    os.kill(gpg_pid, signal.SIGKILL)
        except subp.ProcessExecutionError as e:
            LOG.warning("Failed to clean up gpg process: %s", e)

Batal