Module gnupg.handlers
View Source
from __future__ import annotations
from .add_subkey_handler import AddSubkeyHandler
from .auto_locate_key_handler import AutoLocateKeyHandler
from .crypt_handler import CryptHandler
from .delete_result_handler import DeleteResultHandler
from .export_result_handler import ExportResultHandler
from .gen_key_handler import GenKeyHandler
from .import_result_handler import ImportResultHandler
from .list_keys_handler import ListKeysHandler
from .scan_keys_handler import ScanKeysHandler
from .search_keys_handler import SearchKeysHandler
from .send_result_handler import SendResultHandler
from .sign_handler import SignHandler
from .status_handler import StatusHandler
from .trust_result_handler import TrustResultHandler
from .verify_handler import VerifyHandler
__all__ = (
"AddSubkeyHandler",
"AutoLocateKeyHandler",
"CryptHandler",
"DeleteResultHandler",
"ExportResultHandler",
"GenKeyHandler",
"ImportResultHandler",
"ListKeysHandler",
"ScanKeysHandler",
"SearchKeysHandler",
"SendResultHandler",
"SignHandler",
"StatusHandler",
"TrustResultHandler",
"VerifyHandler",
)
Sub-modules
- gnupg.handlers.add_subkey_handler
- gnupg.handlers.auto_locate_key_handler
- gnupg.handlers.crypt_handler
- gnupg.handlers.delete_result_handler
- gnupg.handlers.export_result_handler
- gnupg.handlers.gen_key_handler
- gnupg.handlers.helper
- gnupg.handlers.import_result_handler
- gnupg.handlers.list_keys_handler
- gnupg.handlers.scan_keys_handler
- gnupg.handlers.search_keys_handler
- gnupg.handlers.send_result_handler
- gnupg.handlers.sign_handler
- gnupg.handlers.status_handler
- gnupg.handlers.text_handler
- gnupg.handlers.trust_result_handler
- gnupg.handlers.verify_handler
Classes
AddSubkeyHandler
class AddSubkeyHandler(
gpg: 'GPG'
)
This class handles status messages during subkey addition.
View Source
class AddSubkeyHandler(StatusHandler):
"""
This class handles status messages during subkey addition.
"""
returncode = None
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.type = None
self.fingerprint = ""
self.status = None
def __nonzero__(self) -> bool: # pragma: no cover
return bool(self.fingerprint)
__bool__ = __nonzero__
def __str__(self) -> str:
return self.fingerprint
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key == "KEY_CREATED":
(self.type, self.fingerprint) = value.split()
self.status = "ok"
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
Class variables
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key == "KEY_CREATED":
(self.type, self.fingerprint) = value.split()
self.status = "ok"
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
AutoLocateKeyHandler
class AutoLocateKeyHandler(
gpg: 'GPG'
)
This class handles status messages during key auto-locating.
fingerprint: str key_length: int created_at: date email: str email_real_name: str
View Source
class AutoLocateKeyHandler(StatusHandler):
"""
This class handles status messages during key auto-locating.
fingerprint: str
key_length: int
created_at: date
email: str
email_real_name: str
"""
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.fingerprint = None
self.type = None
self.created_at = None
self.email = None
self.email_real_name = None
def handle_status(self, key: str, value: str) -> None:
if key == "IMPORTED":
_, email, display_name = value.split()
self.email = email
self.email_real_name = display_name[1:-1]
elif key == "KEY_CONSIDERED":
self.fingerprint = value.strip().split()[0]
def pub(self, args: list[str]) -> None:
"""
Internal method to handle the 'pub' status message.
`pub` message contains the fingerprint of the public key, its type and its creation date.
"""
def uid(self, args: list[str]) -> None:
local_tz = datetime.now().astimezone().tzinfo
self.created_at = datetime.fromtimestamp(int(args[5]), tz=local_tz)
raw_email_content = args[9]
email, real_name = parseaddr(raw_email_content)
self.email = email
self.email_real_name = real_name
def sub(self, args: list[str]) -> None:
self.key_length = int(args[2])
def fpr(self, args: list[str]) -> None:
# Only store the first fingerprint
self.fingerprint = self.fingerprint or args[9]
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
Methods
fpr
def fpr(
self,
args: 'list[str]'
) -> 'None'
View Source
def fpr(self, args: list[str]) -> None:
# Only store the first fingerprint
self.fingerprint = self.fingerprint or args[9]
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key == "IMPORTED":
_, email, display_name = value.split()
self.email = email
self.email_real_name = display_name[1:-1]
elif key == "KEY_CONSIDERED":
self.fingerprint = value.strip().split()[0]
pub
def pub(
self,
args: 'list[str]'
) -> 'None'
Internal method to handle the 'pub' status message.
pub message contains the fingerprint of the public key, its type and its creation date.
View Source
def pub(self, args: list[str]) -> None:
"""
Internal method to handle the 'pub' status message.
`pub` message contains the fingerprint of the public key, its type and its creation date.
"""
sub
def sub(
self,
args: 'list[str]'
) -> 'None'
View Source
def sub(self, args: list[str]) -> None:
self.key_length = int(args[2])
uid
def uid(
self,
args: 'list[str]'
) -> 'None'
View Source
def uid(self, args: list[str]) -> None:
local_tz = datetime.now().astimezone().tzinfo
self.created_at = datetime.fromtimestamp(int(args[5]), tz=local_tz)
raw_email_content = args[9]
email, real_name = parseaddr(raw_email_content)
self.email = email
self.email_real_name = real_name
CryptHandler
class CryptHandler(
gpg: 'GPG'
)
This class handles status messages during encryption and decryption.
View Source
class CryptHandler(VerifyHandler, TextHandler):
"""
This class handles status messages during encryption and decryption.
"""
def __init__(self, gpg: GPG) -> None:
VerifyHandler.__init__(self, gpg)
self.data = ""
self.ok = False
self.status = ""
self.status_detail = ""
self.key_id = None
def __nonzero__(self) -> bool:
return bool(self.ok)
__bool__ = __nonzero__
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"):
logger.warning("potential problem: %s: %s", key, value)
elif key == "NODATA":
if self.status not in ("decryption failed",):
self.status = "no data was provided"
elif key in (
"NEED_PASSPHRASE",
"BAD_PASSPHRASE",
"GOOD_PASSPHRASE",
"MISSING_PASSPHRASE",
"KEY_NOT_CREATED",
"NEED_PASSPHRASE_PIN",
): # pragma: no cover
self.status = key.replace("_", " ").lower()
elif key == "DECRYPTION_FAILED": # pragma: no cover
if self.status != "no secret key": # don't overwrite more useful message
self.status = "decryption failed"
elif key == "NEED_PASSPHRASE_SYM":
self.status = "need symmetric passphrase"
elif key == "BEGIN_DECRYPTION":
if self.status != "no secret key": # don't overwrite more useful message
self.status = "decryption incomplete"
elif key == "BEGIN_ENCRYPTION":
self.status = "encryption incomplete"
elif key == "DECRYPTION_OKAY":
self.status = "decryption ok"
self.ok = True
elif key == "END_ENCRYPTION":
self.status = "encryption ok"
self.ok = True
elif key == "INV_RECP": # pragma: no cover
if not self.status:
self.status = "invalid recipient"
else:
self.status = f"invalid recipient: {self.status}"
self.status_detail = _determine_invalid_recipient_or_signer(value)
elif key == "KEYEXPIRED": # pragma: no cover
self.status = "key expired"
elif key == "SIG_CREATED": # pragma: no cover
self.status = "sig created"
elif key == "SIGEXPIRED": # pragma: no cover
self.status = "sig expired"
elif key == "ENC_TO": # pragma: no cover
# ENC_TO <long_keyid> <keytype> <keylength>
self.key_id = value.split(" ", 1)[0]
elif key in (
"USERID_HINT",
"GOODMDC",
"END_DECRYPTION",
"CARDCTRL",
"BADMDC",
"SC_OP_FAILURE",
"SC_OP_SUCCESS",
"PINENTRY_LAUNCHED",
):
pass
else:
VerifyHandler.handle_status(self, key, value)
Ancestors (in MRO)
- gnupg.handlers.VerifyHandler
- gnupg.handlers.StatusHandler
- gnupg.handlers.text_handler.TextHandler
Class variables
GPG_ERROR_CODES
GPG_SYSTEM_ERROR_CODES
TRUST_EXPIRED
TRUST_FULLY
TRUST_LEVELS
TRUST_MARGINAL
TRUST_NEVER
TRUST_ULTIMATE
TRUST_UNDEFINED
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"):
logger.warning("potential problem: %s: %s", key, value)
elif key == "NODATA":
if self.status not in ("decryption failed",):
self.status = "no data was provided"
elif key in (
"NEED_PASSPHRASE",
"BAD_PASSPHRASE",
"GOOD_PASSPHRASE",
"MISSING_PASSPHRASE",
"KEY_NOT_CREATED",
"NEED_PASSPHRASE_PIN",
): # pragma: no cover
self.status = key.replace("_", " ").lower()
elif key == "DECRYPTION_FAILED": # pragma: no cover
if self.status != "no secret key": # don't overwrite more useful message
self.status = "decryption failed"
elif key == "NEED_PASSPHRASE_SYM":
self.status = "need symmetric passphrase"
elif key == "BEGIN_DECRYPTION":
if self.status != "no secret key": # don't overwrite more useful message
self.status = "decryption incomplete"
elif key == "BEGIN_ENCRYPTION":
self.status = "encryption incomplete"
elif key == "DECRYPTION_OKAY":
self.status = "decryption ok"
self.ok = True
elif key == "END_ENCRYPTION":
self.status = "encryption ok"
self.ok = True
elif key == "INV_RECP": # pragma: no cover
if not self.status:
self.status = "invalid recipient"
else:
self.status = f"invalid recipient: {self.status}"
self.status_detail = _determine_invalid_recipient_or_signer(value)
elif key == "KEYEXPIRED": # pragma: no cover
self.status = "key expired"
elif key == "SIG_CREATED": # pragma: no cover
self.status = "sig created"
elif key == "SIGEXPIRED": # pragma: no cover
self.status = "sig expired"
elif key == "ENC_TO": # pragma: no cover
# ENC_TO <long_keyid> <keytype> <keylength>
self.key_id = value.split(" ", 1)[0]
elif key in (
"USERID_HINT",
"GOODMDC",
"END_DECRYPTION",
"CARDCTRL",
"BADMDC",
"SC_OP_FAILURE",
"SC_OP_SUCCESS",
"PINENTRY_LAUNCHED",
):
pass
else:
VerifyHandler.handle_status(self, key, value)
DeleteResultHandler
class DeleteResultHandler(
gpg: 'GPG'
)
This class handles status messages during key deletion.
View Source
class DeleteResultHandler(StatusHandler):
"""
This class handles status messages during key deletion.
"""
returncode = None
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.status = "ok"
def __str__(self) -> str:
return self.status
problem_reason = MappingProxyType(
{
"1": "No such key",
"2": "Must delete secret key first",
"3": "Ambiguous specification",
},
)
def handle_status(self, key: str, value: str) -> None:
if key == "DELETE_PROBLEM": # pragma: no cover
self.status = self.problem_reason.get(value, f"Unknown error: {value!r}")
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
def __nonzero__(self) -> bool: # pragma: no cover
return self.status == "ok"
__bool__ = __nonzero__
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
Descendants
- gnupg.handlers.TrustResultHandler
Class variables
problem_reason
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key == "DELETE_PROBLEM": # pragma: no cover
self.status = self.problem_reason.get(value, f"Unknown error: {value!r}")
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
ExportResultHandler
class ExportResultHandler(
gpg: 'GPG'
)
This class handles status messages during key export.
View Source
class ExportResultHandler(GenKeyHandler):
"""
This class handles status messages during key export.
"""
# For now, just use an existing class to base it on - if needed, we
# can override handle_status for more specific message handling.
def handle_status(self, key: str, value: str) -> None:
if key in ("EXPORTED", "EXPORT_RES"):
pass
else:
super().handle_status(key, value)
Ancestors (in MRO)
- gnupg.handlers.GenKeyHandler
- gnupg.handlers.StatusHandler
Class variables
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key in ("EXPORTED", "EXPORT_RES"):
pass
else:
super().handle_status(key, value)
GenKeyHandler
class GenKeyHandler(
gpg: 'GPG'
)
This class handles status messages during key generation.
View Source
class GenKeyHandler(StatusHandler):
"""
This class handles status messages during key generation.
"""
returncode = None
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.type = None
self.fingerprint = ""
self.status = None
def __nonzero__(self) -> bool: # pragma: no cover
return bool(self.fingerprint)
__bool__ = __nonzero__
def __str__(self) -> str: # pragma: no cover
return self.fingerprint
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key == "KEY_CREATED":
parts = value.split()
(self.type, self.fingerprint) = parts[:2]
self.status = "ok"
elif key == "KEY_NOT_CREATED":
self.status = key.replace("_", " ").lower()
elif key in ("PROGRESS", "GOOD_PASSPHRASE"): # pragma: no cover
pass
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
Descendants
- gnupg.handlers.ExportResultHandler
Class variables
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key == "KEY_CREATED":
parts = value.split()
(self.type, self.fingerprint) = parts[:2]
self.status = "ok"
elif key == "KEY_NOT_CREATED":
self.status = key.replace("_", " ").lower()
elif key in ("PROGRESS", "GOOD_PASSPHRASE"): # pragma: no cover
pass
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
ImportResultHandler
class ImportResultHandler(
gpg: 'GPG'
)
This class handles status messages during key import.
View Source
class ImportResultHandler(StatusHandler):
"""
This class handles status messages during key import.
"""
counts = """count no_user_id imported imported_rsa unchanged n_uids n_subk n_sigs n_revoc sec_read sec_imported
sec_dups not_imported""".split()
returncode = None
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.results = []
self.fingerprints = []
for result in self.counts:
setattr(self, result, 0)
def __nonzero__(self) -> bool:
return bool(not self.not_imported and self.fingerprints)
__bool__ = __nonzero__
ok_reason = MappingProxyType(
{
"0": "Not actually changed",
"1": "Entirely new key",
"2": "New user IDs",
"4": "New signatures",
"8": "New subkeys",
"16": "Contains private key",
},
)
problem_reason = MappingProxyType(
{
"0": "No specific reason given",
"1": "Invalid Certificate",
"2": "Issuer Certificate missing",
"3": "Certificate Chain too long",
"4": "Error storing certificate",
},
)
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key in ("IMPORTED", "KEY_CONSIDERED"):
# this duplicates info we already see in import_ok & import_problem
pass
elif key == "NODATA": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "No valid data found"})
elif key == "IMPORT_OK":
reason, fingerprint = value.split()
reasons = []
for code, text in list(self.ok_reason.items()):
if int(reason) | int(code) == int(reason):
reasons.append(text)
reasontext = "\n".join(reasons) + "\n"
self.results.append({"fingerprint": fingerprint, "ok": reason, "text": reasontext})
self.fingerprints.append(fingerprint)
elif key == "IMPORT_PROBLEM": # pragma: no cover
try:
reason, fingerprint = value.split()
except (AttributeError, Exception):
reason = value
fingerprint = "<unknown>"
self.results.append({"fingerprint": fingerprint, "problem": reason, "text": self.problem_reason[reason]})
elif key == "IMPORT_RES":
import_res = value.split()
for i, count in enumerate(self.counts):
setattr(self, count, int(import_res[i]))
elif key == "KEYEXPIRED": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "Key expired"})
elif key == "SIGEXPIRED": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "Signature expired"})
elif key == "FAILURE": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "Other failure"})
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
def summary(self) -> str:
"""
Return a summary indicating how many keys were imported and how many were not imported.
"""
result = []
result.append("%d imported" % self.imported)
if self.not_imported: # pragma: no cover
result.append("%d not imported" % self.not_imported)
return ", ".join(result)
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
Class variables
counts
ok_reason
problem_reason
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key in ("IMPORTED", "KEY_CONSIDERED"):
# this duplicates info we already see in import_ok & import_problem
pass
elif key == "NODATA": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "No valid data found"})
elif key == "IMPORT_OK":
reason, fingerprint = value.split()
reasons = []
for code, text in list(self.ok_reason.items()):
if int(reason) | int(code) == int(reason):
reasons.append(text)
reasontext = "\n".join(reasons) + "\n"
self.results.append({"fingerprint": fingerprint, "ok": reason, "text": reasontext})
self.fingerprints.append(fingerprint)
elif key == "IMPORT_PROBLEM": # pragma: no cover
try:
reason, fingerprint = value.split()
except (AttributeError, Exception):
reason = value
fingerprint = "<unknown>"
self.results.append({"fingerprint": fingerprint, "problem": reason, "text": self.problem_reason[reason]})
elif key == "IMPORT_RES":
import_res = value.split()
for i, count in enumerate(self.counts):
setattr(self, count, int(import_res[i]))
elif key == "KEYEXPIRED": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "Key expired"})
elif key == "SIGEXPIRED": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "Signature expired"})
elif key == "FAILURE": # pragma: no cover
self.results.append({"fingerprint": None, "problem": "0", "text": "Other failure"})
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
summary
def summary(
self
) -> 'str'
Return a summary indicating how many keys were imported and how many were not imported.
View Source
def summary(self) -> str:
"""
Return a summary indicating how many keys were imported and how many were not imported.
"""
result = []
result.append("%d imported" % self.imported)
if self.not_imported: # pragma: no cover
result.append("%d not imported" % self.not_imported)
return ", ".join(result)
ListKeysHandler
class ListKeysHandler(
gpg: 'GPG'
)
This class handles status messages during listing keys and signatures.
Handle pub and uid (relating the latter to the former).
We don't care about (info from GnuPG DETAILS file):
crt = X.509 certificate crs = X.509 certificate and private key available uat = user attribute (same as user id except for field 10). sig = signature rev = revocation signature pkd = public key data (special field format, see below) grp = reserved for gpgsm rvk = revocation key
View Source
class ListKeysHandler(SearchKeysHandler):
"""
This class handles status messages during listing keys and signatures.
Handle pub and uid (relating the latter to the former).
We don't care about (info from GnuPG DETAILS file):
crt = X.509 certificate
crs = X.509 certificate and private key available
uat = user attribute (same as user id except for field 10).
sig = signature
rev = revocation signature
pkd = public key data (special field format, see below)
grp = reserved for gpgsm
rvk = revocation key
"""
UID_INDEX = 9
FIELDS = (
"type trust length algo keyid date expires dummy ownertrust uid sig"
" cap issuer flag token hash curve compliance updated origin keygrip"
).split()
def __init__(self, gpg: GPG) -> None:
super().__init__(gpg)
self.in_subkey = False
self.key_map = {}
def key(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
if curkey["uid"]: # pragma: no cover
curkey["uids"].append(curkey["uid"])
del curkey["uid"]
curkey["subkeys"] = []
self.append(curkey)
self.in_subkey = False
pub = sec = key
def fpr(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
fp = args[9]
if fp in self.key_map and self.gpg.check_fingerprint_collisions: # pragma: no cover
msg = f"Unexpected fingerprint collision: {fp}"
raise ValueError(msg)
if not self.in_subkey:
self.curkey["fingerprint"] = fp
self.fingerprints.append(fp)
self.key_map[fp] = self.curkey
else:
self.curkey["subkeys"][-1][2] = fp
self.key_map[fp] = self.curkey
def grp(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
grp = args[9]
if not self.in_subkey:
self.curkey["keygrip"] = grp
else:
self.curkey["subkeys"][-1][3] = grp
def _collect_subkey_info(
self,
curkey: dict[
str,
str | list[str] | list[list[str | None]] | dict[str, dict[str, str]] | list[tuple[str, str, str]],
],
args: list[str],
) -> None:
info_map = curkey.setdefault("subkey_info", {})
info = {}
_set_fields(info, self.FIELDS, args)
info_map[args[4]] = info
def sub(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
# See issue #81. We create a dict with more information about
# subkeys, but for backward compatibility reason, have to add it in
# as a separate entry 'subkey_info'
subkey = [args[4], args[11], None, None] # keyid, type, fp, grp
self.curkey["subkeys"].append(subkey)
self._collect_subkey_info(self.curkey, args)
self.in_subkey = True
def ssb(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
subkey = [args[4], None, None, None] # keyid, type, fp, grp
self.curkey["subkeys"].append(subkey)
self._collect_subkey_info(self.curkey, args)
self.in_subkey = True
def sig(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
# keyid, uid, sigclass
self.curkey["sigs"].append((args[4], args[9], args[10]))
Ancestors (in MRO)
- gnupg.handlers.SearchKeysHandler
- gnupg.handlers.StatusHandler
- builtins.list
Descendants
- gnupg.handlers.ScanKeysHandler
Class variables
FIELDS
UID_INDEX
returncode
Methods
append
def append(
self,
object,
/
)
Append object to the end of the list.
clear
def clear(
self,
/
)
Remove all items from list.
copy
def copy(
self,
/
)
Return a shallow copy of the list.
count
def count(
self,
value,
/
)
Return number of occurrences of value.
extend
def extend(
self,
iterable,
/
)
Extend list by appending elements from the iterable.
fpr
def fpr(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def fpr(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
fp = args[9]
if fp in self.key_map and self.gpg.check_fingerprint_collisions: # pragma: no cover
msg = f"Unexpected fingerprint collision: {fp}"
raise ValueError(msg)
if not self.in_subkey:
self.curkey["fingerprint"] = fp
self.fingerprints.append(fp)
self.key_map[fp] = self.curkey
else:
self.curkey["subkeys"][-1][2] = fp
self.key_map[fp] = self.curkey
get_fields
def get_fields(
self,
args: 'list[str]'
) -> 'dict[str, str]'
Internal method used to update the instance from a gpg status message.
View Source
def get_fields(self, args: list[str]) -> dict[str, str]:
"""
Internal method used to update the instance from a `gpg` status message.
"""
result = {}
_set_fields(result, self.FIELDS, args)
result["uids"] = []
result["sigs"] = []
return result
grp
def grp(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def grp(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
grp = args[9]
if not self.in_subkey:
self.curkey["keygrip"] = grp
else:
self.curkey["subkeys"][-1][3] = grp
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None: # pragma: no cover
pass
index
def index(
self,
value,
start=0,
stop=9223372036854775807,
/
)
Return first index of value.
Raises ValueError if the value is not present.
insert
def insert(
self,
index,
object,
/
)
Insert object before index.
key
def key(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def key(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
if curkey["uid"]: # pragma: no cover
curkey["uids"].append(curkey["uid"])
del curkey["uid"]
curkey["subkeys"] = []
self.append(curkey)
self.in_subkey = False
pop
def pop(
self,
index=-1,
/
)
Remove and return item at index (default last).
Raises IndexError if list is empty or index is out of range.
pub
def pub(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def key(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
if curkey["uid"]: # pragma: no cover
curkey["uids"].append(curkey["uid"])
del curkey["uid"]
curkey["subkeys"] = []
self.append(curkey)
self.in_subkey = False
remove
def remove(
self,
value,
/
)
Remove first occurrence of value.
Raises ValueError if the value is not present.
reverse
def reverse(
self,
/
)
Reverse IN PLACE.
sec
def sec(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def key(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
if curkey["uid"]: # pragma: no cover
curkey["uids"].append(curkey["uid"])
del curkey["uid"]
curkey["subkeys"] = []
self.append(curkey)
self.in_subkey = False
sig
def sig(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def sig(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
# keyid, uid, sigclass
self.curkey["sigs"].append((args[4], args[9], args[10]))
sort
def sort(
self,
/,
*,
key=None,
reverse=False
)
Sort the list in ascending order and return None.
The sort is in-place (i.e. the list itself is modified) and stable (i.e. the order of two equal elements is maintained).
If a key function is given, apply it once to each list item and sort them, ascending or descending, according to their function values.
The reverse flag can be set to sort in descending order.
ssb
def ssb(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def ssb(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
subkey = [args[4], None, None, None] # keyid, type, fp, grp
self.curkey["subkeys"].append(subkey)
self._collect_subkey_info(self.curkey, args)
self.in_subkey = True
sub
def sub(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def sub(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
# See issue #81. We create a dict with more information about
# subkeys, but for backward compatibility reason, have to add it in
# as a separate entry 'subkey_info'
subkey = [args[4], args[11], None, None] # keyid, type, fp, grp
self.curkey["subkeys"].append(subkey)
self._collect_subkey_info(self.curkey, args)
self.in_subkey = True
uid
def uid(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def uid(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
uid = args[self.UID_INDEX]
uid = ESCAPE_PATTERN.sub(lambda m: chr(int(m.group(1), 16)), uid)
for k, v in BASIC_ESCAPES.items():
uid = uid.replace(k, v)
self.curkey["uids"].append(uid)
self.uids.append(uid)
ScanKeysHandler
class ScanKeysHandler(
gpg: 'GPG'
)
This class handles status messages during scanning keys.
View Source
class ScanKeysHandler(ListKeysHandler):
"""
This class handles status messages during scanning keys.
"""
def sub(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
# --with-fingerprint --with-colons somehow outputs fewer colons,
# use the last value args[-1] instead of args[11]
subkey = [args[4], args[-1], None, None]
self.curkey["subkeys"].append(subkey)
self._collect_subkey_info(self.curkey, args)
self.in_subkey = True
Ancestors (in MRO)
- gnupg.handlers.ListKeysHandler
- gnupg.handlers.SearchKeysHandler
- gnupg.handlers.StatusHandler
- builtins.list
Class variables
FIELDS
UID_INDEX
returncode
Methods
append
def append(
self,
object,
/
)
Append object to the end of the list.
clear
def clear(
self,
/
)
Remove all items from list.
copy
def copy(
self,
/
)
Return a shallow copy of the list.
count
def count(
self,
value,
/
)
Return number of occurrences of value.
extend
def extend(
self,
iterable,
/
)
Extend list by appending elements from the iterable.
fpr
def fpr(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def fpr(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
fp = args[9]
if fp in self.key_map and self.gpg.check_fingerprint_collisions: # pragma: no cover
msg = f"Unexpected fingerprint collision: {fp}"
raise ValueError(msg)
if not self.in_subkey:
self.curkey["fingerprint"] = fp
self.fingerprints.append(fp)
self.key_map[fp] = self.curkey
else:
self.curkey["subkeys"][-1][2] = fp
self.key_map[fp] = self.curkey
get_fields
def get_fields(
self,
args: 'list[str]'
) -> 'dict[str, str]'
Internal method used to update the instance from a gpg status message.
View Source
def get_fields(self, args: list[str]) -> dict[str, str]:
"""
Internal method used to update the instance from a `gpg` status message.
"""
result = {}
_set_fields(result, self.FIELDS, args)
result["uids"] = []
result["sigs"] = []
return result
grp
def grp(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def grp(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
grp = args[9]
if not self.in_subkey:
self.curkey["keygrip"] = grp
else:
self.curkey["subkeys"][-1][3] = grp
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None: # pragma: no cover
pass
index
def index(
self,
value,
start=0,
stop=9223372036854775807,
/
)
Return first index of value.
Raises ValueError if the value is not present.
insert
def insert(
self,
index,
object,
/
)
Insert object before index.
key
def key(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def key(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
if curkey["uid"]: # pragma: no cover
curkey["uids"].append(curkey["uid"])
del curkey["uid"]
curkey["subkeys"] = []
self.append(curkey)
self.in_subkey = False
pop
def pop(
self,
index=-1,
/
)
Remove and return item at index (default last).
Raises IndexError if list is empty or index is out of range.
pub
def pub(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def key(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
if curkey["uid"]: # pragma: no cover
curkey["uids"].append(curkey["uid"])
del curkey["uid"]
curkey["subkeys"] = []
self.append(curkey)
self.in_subkey = False
remove
def remove(
self,
value,
/
)
Remove first occurrence of value.
Raises ValueError if the value is not present.
reverse
def reverse(
self,
/
)
Reverse IN PLACE.
sec
def sec(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def key(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
if curkey["uid"]: # pragma: no cover
curkey["uids"].append(curkey["uid"])
del curkey["uid"]
curkey["subkeys"] = []
self.append(curkey)
self.in_subkey = False
sig
def sig(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def sig(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
# keyid, uid, sigclass
self.curkey["sigs"].append((args[4], args[9], args[10]))
sort
def sort(
self,
/,
*,
key=None,
reverse=False
)
Sort the list in ascending order and return None.
The sort is in-place (i.e. the list itself is modified) and stable (i.e. the order of two equal elements is maintained).
If a key function is given, apply it once to each list item and sort them, ascending or descending, according to their function values.
The reverse flag can be set to sort in descending order.
ssb
def ssb(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def ssb(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
subkey = [args[4], None, None, None] # keyid, type, fp, grp
self.curkey["subkeys"].append(subkey)
self._collect_subkey_info(self.curkey, args)
self.in_subkey = True
sub
def sub(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def sub(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
# --with-fingerprint --with-colons somehow outputs fewer colons,
# use the last value args[-1] instead of args[11]
subkey = [args[4], args[-1], None, None]
self.curkey["subkeys"].append(subkey)
self._collect_subkey_info(self.curkey, args)
self.in_subkey = True
uid
def uid(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def uid(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
uid = args[self.UID_INDEX]
uid = ESCAPE_PATTERN.sub(lambda m: chr(int(m.group(1), 16)), uid)
for k, v in BASIC_ESCAPES.items():
uid = uid.replace(k, v)
self.curkey["uids"].append(uid)
self.uids.append(uid)
SearchKeysHandler
class SearchKeysHandler(
gpg: 'GPG'
)
This class handles status messages during key search.
View Source
class SearchKeysHandler(StatusHandler, list):
"""
This class handles status messages during key search.
"""
# Handle pub and uid (relating the latter to the former).
# Don't care about the rest
UID_INDEX = 1
FIELDS = "type keyid algo length date expires".split()
returncode = None
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.curkey = None
self.fingerprints = []
self.uids = []
def get_fields(self, args: list[str]) -> dict[str, str]:
"""
Internal method used to update the instance from a `gpg` status message.
"""
result = {}
_set_fields(result, self.FIELDS, args)
result["uids"] = []
result["sigs"] = []
return result
def pub(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
self.append(curkey)
def uid(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
uid = args[self.UID_INDEX]
uid = ESCAPE_PATTERN.sub(lambda m: chr(int(m.group(1), 16)), uid)
for k, v in BASIC_ESCAPES.items():
uid = uid.replace(k, v)
self.curkey["uids"].append(uid)
self.uids.append(uid)
def handle_status(self, key: str, value: str) -> None: # pragma: no cover
pass
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
- builtins.list
Descendants
- gnupg.handlers.ListKeysHandler
Class variables
FIELDS
UID_INDEX
returncode
Methods
append
def append(
self,
object,
/
)
Append object to the end of the list.
clear
def clear(
self,
/
)
Remove all items from list.
copy
def copy(
self,
/
)
Return a shallow copy of the list.
count
def count(
self,
value,
/
)
Return number of occurrences of value.
extend
def extend(
self,
iterable,
/
)
Extend list by appending elements from the iterable.
get_fields
def get_fields(
self,
args: 'list[str]'
) -> 'dict[str, str]'
Internal method used to update the instance from a gpg status message.
View Source
def get_fields(self, args: list[str]) -> dict[str, str]:
"""
Internal method used to update the instance from a `gpg` status message.
"""
result = {}
_set_fields(result, self.FIELDS, args)
result["uids"] = []
result["sigs"] = []
return result
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None: # pragma: no cover
pass
index
def index(
self,
value,
start=0,
stop=9223372036854775807,
/
)
Return first index of value.
Raises ValueError if the value is not present.
insert
def insert(
self,
index,
object,
/
)
Insert object before index.
pop
def pop(
self,
index=-1,
/
)
Remove and return item at index (default last).
Raises IndexError if list is empty or index is out of range.
pub
def pub(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def pub(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
self.curkey = curkey = self.get_fields(args)
self.append(curkey)
remove
def remove(
self,
value,
/
)
Remove first occurrence of value.
Raises ValueError if the value is not present.
reverse
def reverse(
self,
/
)
Reverse IN PLACE.
sort
def sort(
self,
/,
*,
key=None,
reverse=False
)
Sort the list in ascending order and return None.
The sort is in-place (i.e. the list itself is modified) and stable (i.e. the order of two equal elements is maintained).
If a key function is given, apply it once to each list item and sort them, ascending or descending, according to their function values.
The reverse flag can be set to sort in descending order.
uid
def uid(
self,
args: 'list[str]'
) -> 'None'
Internal method used to update the instance from a gpg status message.
View Source
def uid(self, args: list[str]) -> None:
"""
Internal method used to update the instance from a `gpg` status message.
"""
uid = args[self.UID_INDEX]
uid = ESCAPE_PATTERN.sub(lambda m: chr(int(m.group(1), 16)), uid)
for k, v in BASIC_ESCAPES.items():
uid = uid.replace(k, v)
self.curkey["uids"].append(uid)
self.uids.append(uid)
SendResultHandler
class SendResultHandler(
gpg: 'GPG'
)
This class handles status messages during key sending.
View Source
class SendResultHandler(StatusHandler):
"""
This class handles status messages during key sending.
"""
returncode = None
def handle_status(self, key: str, value: str) -> None:
logger.debug("SendResult: %s: %s", key, value)
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
Class variables
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
logger.debug("SendResult: %s: %s", key, value)
SignHandler
class SignHandler(
gpg: 'GPG'
)
This class handles status messages during signing.
View Source
class SignHandler(StatusHandler, TextHandler):
"""
This class handles status messages during signing.
"""
returncode = None
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.type = None
self.hash_algo = None
self.fingerprint = None
self.status = None
self.status_detail = None
self.key_id = None
self.username = None
def __nonzero__(self) -> bool:
return self.fingerprint is not None
__bool__ = __nonzero__
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR", "FAILURE"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key in ("KEYEXPIRED", "SIGEXPIRED"): # pragma: no cover
self.status = "key expired"
elif key == "KEYREVOKED": # pragma: no cover
self.status = "key revoked"
elif key == "SIG_CREATED":
(self.type, algo, self.hash_algo, cls, self.timestamp, self.fingerprint) = value.split()
self.status = "signature created"
elif key == "USERID_HINT": # pragma: no cover
self.key_id, self.username = value.split(" ", 1)
elif key == "BAD_PASSPHRASE": # pragma: no cover
self.status = "bad passphrase"
elif key in ("INV_SGNR", "INV_RECP"): # pragma: no cover
# INV_RECP is returned in older versions
if not self.status:
self.status = "invalid signer"
else:
self.status = f"invalid signer: {self.status}"
self.status_detail = _determine_invalid_recipient_or_signer(value)
elif key in ("NEED_PASSPHRASE", "GOOD_PASSPHRASE", "BEGIN_SIGNING"):
pass
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
- gnupg.handlers.text_handler.TextHandler
Class variables
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key in ("WARNING", "ERROR", "FAILURE"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key in ("KEYEXPIRED", "SIGEXPIRED"): # pragma: no cover
self.status = "key expired"
elif key == "KEYREVOKED": # pragma: no cover
self.status = "key revoked"
elif key == "SIG_CREATED":
(self.type, algo, self.hash_algo, cls, self.timestamp, self.fingerprint) = value.split()
self.status = "signature created"
elif key == "USERID_HINT": # pragma: no cover
self.key_id, self.username = value.split(" ", 1)
elif key == "BAD_PASSPHRASE": # pragma: no cover
self.status = "bad passphrase"
elif key in ("INV_SGNR", "INV_RECP"): # pragma: no cover
# INV_RECP is returned in older versions
if not self.status:
self.status = "invalid signer"
else:
self.status = f"invalid signer: {self.status}"
self.status_detail = _determine_invalid_recipient_or_signer(value)
elif key in ("NEED_PASSPHRASE", "GOOD_PASSPHRASE", "BEGIN_SIGNING"):
pass
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
StatusHandler
class StatusHandler(
gpg: 'GPG'
)
The base class for handling status messages from gpg.
View Source
class StatusHandler:
"""
The base class for handling status messages from `gpg`.
"""
def __init__(self, gpg: GPG) -> None:
"""
Initialize an instance.
Args:
gpg (GPG): The :class:`GPG` instance in use.
"""
self.gpg = gpg
self.data: str | None = None
def handle_status(self, key: str, value: str) -> NoReturn:
"""
Handle status messages from the `gpg` child process. These are lines of the format
[GNUPG:] <key> <value>
Args:
key (str): Identifies what the status message is.
value (str): Identifies additional data, which differs depending on the key.
"""
raise NotImplementedError
Descendants
- gnupg.handlers.AddSubkeyHandler
- gnupg.handlers.AutoLocateKeyHandler
- gnupg.handlers.VerifyHandler
- gnupg.handlers.DeleteResultHandler
- gnupg.handlers.GenKeyHandler
- gnupg.handlers.ImportResultHandler
- gnupg.handlers.SearchKeysHandler
- gnupg.handlers.SendResultHandler
- gnupg.handlers.SignHandler
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'NoReturn'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> NoReturn:
"""
Handle status messages from the `gpg` child process. These are lines of the format
[GNUPG:] <key> <value>
Args:
key (str): Identifies what the status message is.
value (str): Identifies additional data, which differs depending on the key.
"""
raise NotImplementedError
TrustResultHandler
class TrustResultHandler(
gpg: 'GPG'
)
This class handles status messages during key trust setting.
View Source
class TrustResultHandler(DeleteResultHandler):
"""
This class handles status messages during key trust setting.
"""
Ancestors (in MRO)
- gnupg.handlers.DeleteResultHandler
- gnupg.handlers.StatusHandler
Class variables
problem_reason
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
if key == "DELETE_PROBLEM": # pragma: no cover
self.status = self.problem_reason.get(value, f"Unknown error: {value!r}")
else: # pragma: no cover
logger.debug("message ignored: %s, %s", key, value)
VerifyHandler
class VerifyHandler(
gpg: 'GPG'
)
This class handles status messages during signature verificaton.
View Source
class VerifyHandler(StatusHandler):
"""
This class handles status messages during signature verificaton.
"""
TRUST_EXPIRED = 0
TRUST_UNDEFINED = 1
TRUST_NEVER = 2
TRUST_MARGINAL = 3
TRUST_FULLY = 4
TRUST_ULTIMATE = 5
TRUST_LEVELS = MappingProxyType(
{
"TRUST_EXPIRED": TRUST_EXPIRED,
"TRUST_UNDEFINED": TRUST_UNDEFINED,
"TRUST_NEVER": TRUST_NEVER,
"TRUST_MARGINAL": TRUST_MARGINAL,
"TRUST_FULLY": TRUST_FULLY,
"TRUST_ULTIMATE": TRUST_ULTIMATE,
},
)
# for now, just the most common error codes. This can be expanded as and
# when reports come in of other errors.
GPG_SYSTEM_ERROR_CODES = MappingProxyType(
{
1: "permission denied",
35: "file exists",
81: "file not found",
97: "not a directory",
},
)
GPG_ERROR_CODES = MappingProxyType(
{
11: "incorrect passphrase",
},
)
returncode = None
def __init__(self, gpg: GPG) -> None:
StatusHandler.__init__(self, gpg)
self.valid = False
self.fingerprint = self.creation_date = self.timestamp = None
self.signature_id = self.key_id = None
self.username = None
self.key_id = None
self.key_status = None
self.status = None
self.pubkey_fingerprint = None
self.expire_timestamp = None
self.sig_timestamp = None
self.trust_text = None
self.trust_level = None
self.sig_info = {}
self.problems = []
def __nonzero__(self) -> bool: # pragma: no cover
return self.valid
__bool__ = __nonzero__
def handle_status(self, key: str, value: str) -> None:
def update_sig_info(**kwargs: object) -> None:
sig_id = self.signature_id
if sig_id:
info = self.sig_info[sig_id]
info.update(kwargs)
else:
logger.debug("Ignored due to missing sig iD: %s", kwargs)
if key in self.TRUST_LEVELS:
self.trust_text = key
self.trust_level = self.TRUST_LEVELS[key]
update_sig_info(trust_level=self.trust_level, trust_text=self.trust_text)
# See Issue #214. Once we see this, we're done with the signature just seen.
# Zap the signature ID, because we don't see a SIG_ID unless we have a new
# good signature.
self.signature_id = None
elif key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key == "BADSIG": # pragma: no cover
self.valid = False
self.status = "signature bad"
self.key_id, self.username = value.split(None, 1)
self.problems.append({"status": self.status, "keyid": self.key_id, "user": self.username})
update_sig_info(keyid=self.key_id, username=self.username, status=self.status)
elif key == "ERRSIG": # pragma: no cover
self.valid = False
parts = value.split()
(self.key_id, algo, hash_algo, cls, self.timestamp) = parts[:5]
# Since GnuPG 2.2.7, a fingerprint is tacked on
if len(parts) >= 7: # noqa: PLR2004
self.fingerprint = parts[6]
self.status = "signature error"
update_sig_info(
keyid=self.key_id,
timestamp=self.timestamp,
fingerprint=self.fingerprint,
status=self.status,
)
self.problems.append(
{
"status": self.status,
"keyid": self.key_id,
"timestamp": self.timestamp,
"fingerprint": self.fingerprint,
},
)
elif key == "EXPSIG": # pragma: no cover
self.valid = False
self.status = "signature expired"
self.key_id, self.username = value.split(None, 1)
update_sig_info(keyid=self.key_id, username=self.username, status=self.status)
self.problems.append({"status": self.status, "keyid": self.key_id, "user": self.username})
elif key == "GOODSIG":
self.valid = True
self.status = "signature good"
self.key_id, self.username = value.split(None, 1)
update_sig_info(keyid=self.key_id, username=self.username, status=self.status)
elif key == "VALIDSIG":
parts = value.split()
fingerprint, creation_date, sig_ts, expire_ts = parts[:4]
(self.fingerprint, self.creation_date, self.sig_timestamp, self.expire_timestamp) = (
fingerprint,
creation_date,
sig_ts,
expire_ts,
)
# may be different if signature is made with a subkey
if len(parts) >= 10: # noqa: PLR2004
self.pubkey_fingerprint = parts[9]
self.status = "signature valid"
update_sig_info(
fingerprint=fingerprint,
creation_date=creation_date,
timestamp=sig_ts,
expiry=expire_ts,
pubkey_fingerprint=self.pubkey_fingerprint,
status=self.status,
)
elif key == "SIG_ID":
sig_id, creation_date, timestamp = value.split()
self.sig_info[sig_id] = {"creation_date": creation_date, "timestamp": timestamp}
(self.signature_id, self.creation_date, self.timestamp) = (sig_id, creation_date, timestamp)
elif key == "NO_PUBKEY": # pragma: no cover
self.valid = False
self.key_id = value
self.status = "no public key"
self.problems.append({"status": self.status, "keyid": self.key_id})
elif key == "NO_SECKEY": # pragma: no cover
self.valid = False
self.key_id = value
self.status = "no secret key"
self.problems.append({"status": self.status, "keyid": self.key_id})
elif key in ("EXPKEYSIG", "REVKEYSIG"): # pragma: no cover
# signed with expired or revoked key
self.valid = False
self.key_id = value.split()[0]
if key == "EXPKEYSIG":
self.key_status = "signing key has expired"
else:
self.key_status = "signing key was revoked"
self.status = self.key_status
update_sig_info(status=self.status, keyid=self.key_id)
self.problems.append({"status": self.status, "keyid": self.key_id})
elif key in ("UNEXPECTED", "FAILURE"): # pragma: no cover
self.valid = False
if key == "UNEXPECTED":
self.status = "unexpected data"
else:
# N.B. there might be other reasons. For example, if an output
# file can't be created - /dev/null/foo will lead to a
# "not a directory" error, but which is not sent as a status
# message with the [GNUPG:] prefix. Similarly if you try to
# write to "/etc/foo" as a non-root user, a "permission denied"
# error will be sent as a non-status message.
message = f"error - {value}"
operation, code = value.rsplit(" ", 1)
if code.isdigit():
code = int(code) & 0xFFFFFF # lose the error source
if self.gpg.error_map and code in self.gpg.error_map:
message = f"{operation}: {self.gpg.error_map[code]}"
else:
system_error = bool(code & 0x8000)
code = code & 0x7FFF
mapping = self.GPG_SYSTEM_ERROR_CODES if system_error else self.GPG_ERROR_CODES
if code in mapping:
message = f"{operation}: {mapping[code]}"
if not self.status:
self.status = message
elif key == "NODATA": # pragma: no cover
# See issue GH-191
self.valid = False
self.status = "signature expected but not found"
elif key in ("DECRYPTION_INFO", "PLAINTEXT", "PLAINTEXT_LENGTH", "BEGIN_SIGNING", "KEY_CONSIDERED"):
pass
elif key in ("NEWSIG",):
# Only sent in gpg2. Clear any signature ID, to be set by a following SIG_ID
self.signature_id = None
else: # pragma: no cover
logger.debug("message ignored: %r, %r", key, value)
Ancestors (in MRO)
- gnupg.handlers.StatusHandler
Descendants
- gnupg.handlers.CryptHandler
Class variables
GPG_ERROR_CODES
GPG_SYSTEM_ERROR_CODES
TRUST_EXPIRED
TRUST_FULLY
TRUST_LEVELS
TRUST_MARGINAL
TRUST_NEVER
TRUST_ULTIMATE
TRUST_UNDEFINED
returncode
Methods
handle_status
def handle_status(
self,
key: 'str',
value: 'str'
) -> 'None'
Handle status messages from the gpg child process. These are lines of the format
[GNUPG:]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| key | str | Identifies what the status message is. | None |
| value | str | Identifies additional data, which differs depending on the key. | None |
View Source
def handle_status(self, key: str, value: str) -> None:
def update_sig_info(**kwargs: object) -> None:
sig_id = self.signature_id
if sig_id:
info = self.sig_info[sig_id]
info.update(kwargs)
else:
logger.debug("Ignored due to missing sig iD: %s", kwargs)
if key in self.TRUST_LEVELS:
self.trust_text = key
self.trust_level = self.TRUST_LEVELS[key]
update_sig_info(trust_level=self.trust_level, trust_text=self.trust_text)
# See Issue #214. Once we see this, we're done with the signature just seen.
# Zap the signature ID, because we don't see a SIG_ID unless we have a new
# good signature.
self.signature_id = None
elif key in ("WARNING", "ERROR"): # pragma: no cover
logger.warning("potential problem: %s: %s", key, value)
elif key == "BADSIG": # pragma: no cover
self.valid = False
self.status = "signature bad"
self.key_id, self.username = value.split(None, 1)
self.problems.append({"status": self.status, "keyid": self.key_id, "user": self.username})
update_sig_info(keyid=self.key_id, username=self.username, status=self.status)
elif key == "ERRSIG": # pragma: no cover
self.valid = False
parts = value.split()
(self.key_id, algo, hash_algo, cls, self.timestamp) = parts[:5]
# Since GnuPG 2.2.7, a fingerprint is tacked on
if len(parts) >= 7: # noqa: PLR2004
self.fingerprint = parts[6]
self.status = "signature error"
update_sig_info(
keyid=self.key_id,
timestamp=self.timestamp,
fingerprint=self.fingerprint,
status=self.status,
)
self.problems.append(
{
"status": self.status,
"keyid": self.key_id,
"timestamp": self.timestamp,
"fingerprint": self.fingerprint,
},
)
elif key == "EXPSIG": # pragma: no cover
self.valid = False
self.status = "signature expired"
self.key_id, self.username = value.split(None, 1)
update_sig_info(keyid=self.key_id, username=self.username, status=self.status)
self.problems.append({"status": self.status, "keyid": self.key_id, "user": self.username})
elif key == "GOODSIG":
self.valid = True
self.status = "signature good"
self.key_id, self.username = value.split(None, 1)
update_sig_info(keyid=self.key_id, username=self.username, status=self.status)
elif key == "VALIDSIG":
parts = value.split()
fingerprint, creation_date, sig_ts, expire_ts = parts[:4]
(self.fingerprint, self.creation_date, self.sig_timestamp, self.expire_timestamp) = (
fingerprint,
creation_date,
sig_ts,
expire_ts,
)
# may be different if signature is made with a subkey
if len(parts) >= 10: # noqa: PLR2004
self.pubkey_fingerprint = parts[9]
self.status = "signature valid"
update_sig_info(
fingerprint=fingerprint,
creation_date=creation_date,
timestamp=sig_ts,
expiry=expire_ts,
pubkey_fingerprint=self.pubkey_fingerprint,
status=self.status,
)
elif key == "SIG_ID":
sig_id, creation_date, timestamp = value.split()
self.sig_info[sig_id] = {"creation_date": creation_date, "timestamp": timestamp}
(self.signature_id, self.creation_date, self.timestamp) = (sig_id, creation_date, timestamp)
elif key == "NO_PUBKEY": # pragma: no cover
self.valid = False
self.key_id = value
self.status = "no public key"
self.problems.append({"status": self.status, "keyid": self.key_id})
elif key == "NO_SECKEY": # pragma: no cover
self.valid = False
self.key_id = value
self.status = "no secret key"
self.problems.append({"status": self.status, "keyid": self.key_id})
elif key in ("EXPKEYSIG", "REVKEYSIG"): # pragma: no cover
# signed with expired or revoked key
self.valid = False
self.key_id = value.split()[0]
if key == "EXPKEYSIG":
self.key_status = "signing key has expired"
else:
self.key_status = "signing key was revoked"
self.status = self.key_status
update_sig_info(status=self.status, keyid=self.key_id)
self.problems.append({"status": self.status, "keyid": self.key_id})
elif key in ("UNEXPECTED", "FAILURE"): # pragma: no cover
self.valid = False
if key == "UNEXPECTED":
self.status = "unexpected data"
else:
# N.B. there might be other reasons. For example, if an output
# file can't be created - /dev/null/foo will lead to a
# "not a directory" error, but which is not sent as a status
# message with the [GNUPG:] prefix. Similarly if you try to
# write to "/etc/foo" as a non-root user, a "permission denied"
# error will be sent as a non-status message.
message = f"error - {value}"
operation, code = value.rsplit(" ", 1)
if code.isdigit():
code = int(code) & 0xFFFFFF # lose the error source
if self.gpg.error_map and code in self.gpg.error_map:
message = f"{operation}: {self.gpg.error_map[code]}"
else:
system_error = bool(code & 0x8000)
code = code & 0x7FFF
mapping = self.GPG_SYSTEM_ERROR_CODES if system_error else self.GPG_ERROR_CODES
if code in mapping:
message = f"{operation}: {mapping[code]}"
if not self.status:
self.status = message
elif key == "NODATA": # pragma: no cover
# See issue GH-191
self.valid = False
self.status = "signature expected but not found"
elif key in ("DECRYPTION_INFO", "PLAINTEXT", "PLAINTEXT_LENGTH", "BEGIN_SIGNING", "KEY_CONSIDERED"):
pass
elif key in ("NEWSIG",):
# Only sent in gpg2. Clear any signature ID, to be set by a following SIG_ID
self.signature_id = None
else: # pragma: no cover
logger.debug("message ignored: %r, %r", key, value)