libervia-backend: libervia/backend/plugins/plugin_xep

comparison libervia/backend/plugins/plugin_xep_0373.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	b53b6dc1f929
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 "GPGPublicKey",
 "GPGSecretKey",
 "GPGProvider",
 "PublicKeyMetadata",
 "gpg_provider",
-"TrustLevel"
+"TrustLevel",
 ]
 log = cast(Logger, getLogger(__name__))  # type: ignore[no-untyped-call]
 PLUGIN_INFO = {
 C.PI_NAME: "XEP-0373",
 C.PI_IMPORT_NAME: "XEP-0373",
 C.PI_TYPE: "SEC",
-C.PI_PROTOCOLS: [ "XEP-0373" ],
+C.PI_PROTOCOLS: ["XEP-0373"],
-C.PI_DEPENDENCIES: [ "XEP-0060", "XEP-0163" ],
+C.PI_DEPENDENCIES: ["XEP-0060", "XEP-0163"],
 C.PI_RECOMMENDATIONS: [],
 C.PI_MAIN: "XEP_0373",
 C.PI_HANDLER: "no",
 C.PI_DESCRIPTION: D_("Implementation of OpenPGP for XMPP"),
 }
 of the public keys is available.
 """
 @abstractmethod
 def verify_detached(
-self,
+self, data: bytes, signature: bytes, public_keys: Set[GPGPublicKey]
-data: bytes,
-signature: bytes,
-public_keys: Set[GPGPublicKey]
 ) -> None:
 """Verify signed data, where the signature was created detached from the data.
 OpenPGP's ASCII Armor is not used.
 @abstractmethod
 def encrypt(
 self,
 plaintext: bytes,
 public_keys: Set[GPGPublicKey],
-signing_keys: Optional[Set[GPGSecretKey]] = None
+signing_keys: Optional[Set[GPGSecretKey]] = None,
 ) -> bytes:
 """Encrypt and optionally sign some data.
 OpenPGP's ASCII Armor is not used.
 @abstractmethod
 def decrypt(
 self,
 ciphertext: bytes,
 secret_keys: Set[GPGSecretKey],
-public_keys: Optional[Set[GPGPublicKey]] = None
+public_keys: Optional[Set[GPGPublicKey]] = None,
 ) -> bytes:
 """Decrypt and optionally verify some data.
 OpenPGP's ASCII Armor is not used.
 def decrypt_symmetrically(self, ciphertext: bytes, password: str) -> bytes:
 with gpg.Context(home_dir=self.__home_dir) as c:
 try:
 plaintext, __, __ = c.decrypt(
-ciphertext,
+ciphertext, passphrase=password, verify=False
-passphrase=password,
-verify=False
 )
 except gpg.errors.GPGMEError as e:
 # TODO: Find out what kind of error is raised if the password is wrong and
 # re-raise it as DecryptionFailed instead.
 raise GPGProviderError("Internal GPGME error") from e
 )
 return data
 def verify_detached(
-self,
+self, data: bytes, signature: bytes, public_keys: Set[GPGPublicKey]
-data: bytes,
-signature: bytes,
-public_keys: Set[GPGPublicKey]
 ) -> None:
 with gpg.Context(home_dir=self.__home_dir) as c:
 try:
 __, result = c.verify(data, signature=signature)
 except gpg.errors.GPGMEError as e:
 def encrypt(
 self,
 plaintext: bytes,
 public_keys: Set[GPGPublicKey],
-signing_keys: Optional[Set[GPGSecretKey]] = None
+signing_keys: Optional[Set[GPGSecretKey]] = None,
 ) -> bytes:
 recipients = []
 for public_key in public_keys:
 assert isinstance(public_key, GPGME_GPGPublicKey)
 ciphertext, __, __ = c.encrypt(
 plaintext,
 recipients=recipients,
 sign=sign,
 always_trust=True,
-add_encrypt_to=True
+add_encrypt_to=True,
 )
 except gpg.errors.GPGMEError as e:
 raise GPGProviderError("Internal GPGME error") from e
 except gpg.errors.InvalidRecipients as e:
 raise GPGProviderError(
 def decrypt(
 self,
 ciphertext: bytes,
 secret_keys: Set[GPGSecretKey],
-public_keys: Optional[Set[GPGPublicKey]] = None
+public_keys: Optional[Set[GPGPublicKey]] = None,
 ) -> bytes:
 verify = public_keys is not None
 with gpg.Context(home_dir=self.__home_dir) as c:
 try:
-plaintext, result, verify_result = c.decrypt(
+plaintext, result, verify_result = c.decrypt(ciphertext, verify=verify)
-ciphertext,
-verify=verify
-)
 except gpg.errors.GPGMEError as e:
 raise GPGProviderError("Internal GPGME error") from e
 except gpg.UnsupportedAlgorithm as e:
 raise DecryptionFailed("Unsupported algorithm") from e
 def list_public_keys(self, user_id: str) -> Set[GPGPublicKey]:
 with gpg.Context(home_dir=self.__home_dir) as c:
 try:
 return {
 GPGME_GPGPublicKey(key)
-for key
+for key in c.keylist(pattern=user_id, secret=False)
-in c.keylist(pattern=user_id, secret=False)
 }
 except gpg.errors.GPGMEError as e:
 raise GPGProviderError("Internal GPGME error") from e
 def list_secret_keys(self, user_id: str) -> Set[GPGSecretKey]:
 with gpg.Context(home_dir=self.__home_dir) as c:
 try:
 return {
 GPGME_GPGSecretKey(GPGME_GPGPublicKey(key))
-for key
+for key in c.keylist(pattern=user_id, secret=True)
-in c.keylist(pattern=user_id, secret=True)
 }
 except gpg.errors.GPGMEError as e:
 raise GPGProviderError("Internal GPGME error") from e
 def can_sign(self, public_key: GPGPublicKey) -> bool:
 expires=False,
 sign=True,
 encrypt=True,
 certify=False,
 authenticate=False,
-force=True
+force=True,
 )
 key_obj = c.get_key(result.fpr, secret=True)
 except gpg.errors.GPGMEError as e:
 raise GPGProviderError("Internal GPGME error") from e
 class PublicKeyMetadata(NamedTuple):
 """
 Metadata about a published public key.
 """
 fingerprint: str
 timestamp: datetime
 def to_dict(self) -> dict:
 # Convert the instance to a dictionary and handle datetime serialization
 data = self._asdict()
-data['timestamp'] = self.timestamp.isoformat()
+data["timestamp"] = self.timestamp.isoformat()
 return data
 @staticmethod
-def from_dict(data: dict) -> 'PublicKeyMetadata':
+def from_dict(data: dict) -> "PublicKeyMetadata":
 # Load a serialised dictionary
-data['timestamp'] = datetime.fromisoformat(data['timestamp'])
+data["timestamp"] = datetime.fromisoformat(data["timestamp"])
 return PublicKeyMetadata(**data)
 @enum.unique
 class TrustLevel(enum.Enum):
 BLINDLY_TRUSTED: str = "BLINDLY_TRUSTED"
 UNDECIDED: str = "UNDECIDED"
 DISTRUSTED: str = "DISTRUSTED"
-OPENPGP_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
+OPENPGP_SCHEMA = xmlschema.XMLSchema(
+"""<?xml version="1.0" encoding="utf8"?>
 <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
 targetNamespace="urn:xmpp:openpgp:0"
 xmlns="urn:xmpp:openpgp:0">
 <xs:element name="openpgp" type="xs:base64Binary"/>
 </xs:schema>
-""")
+"""
+)
 # The following schema needs verion 1.1 of XML Schema, which is not supported by lxml.
 # Luckily, xmlschema exists, which is a clean, well maintained, cross-platform
 # implementation of XML Schema, including version 1.1.
-CONTENT_SCHEMA = xmlschema.XMLSchema11("""<?xml version="1.1" encoding="utf8"?>
+CONTENT_SCHEMA = xmlschema.XMLSchema11(
+"""<?xml version="1.1" encoding="utf8"?>
 <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
 targetNamespace="urn:xmpp:openpgp:0"
 xmlns="urn:xmpp:openpgp:0">
 <xs:element name="signcrypt">
 <xs:any minOccurs="0" maxOccurs="unbounded" processContents="skip"/>
 </xs:sequence>
 </xs:complexType>
 </xs:element>
 </xs:schema>
-""")
+"""
+)
 PUBLIC_KEYS_LIST_NODE = "urn:xmpp:openpgp:0:public-keys"
-PUBLIC_KEYS_LIST_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
+PUBLIC_KEYS_LIST_SCHEMA = xmlschema.XMLSchema(
+"""<?xml version="1.0" encoding="utf8"?>
 <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
 targetNamespace="urn:xmpp:openpgp:0"
 xmlns="urn:xmpp:openpgp:0">
 <xs:element name="public-keys-list">
 <xs:attribute name="v4-fingerprint" type="xs:string"/>
 <xs:attribute name="date" type="xs:dateTime"/>
 </xs:complexType>
 </xs:element>
 </xs:schema>
-""")
+"""
+)
-PUBKEY_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
+PUBKEY_SCHEMA = xmlschema.XMLSchema(
+"""<?xml version="1.0" encoding="utf8"?>
 <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
 targetNamespace="urn:xmpp:openpgp:0"
 xmlns="urn:xmpp:openpgp:0">
 <xs:element name="pubkey">
 </xs:complexType>
 </xs:element>
 <xs:element name="data" type="xs:base64Binary"/>
 </xs:schema>
-""")
+"""
+)
-SECRETKEY_SCHEMA = xmlschema.XMLSchema("""<?xml version="1.0" encoding="utf8"?>
+SECRETKEY_SCHEMA = xmlschema.XMLSchema(
+"""<?xml version="1.0" encoding="utf8"?>
 <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
 targetNamespace="urn:xmpp:openpgp:0"
 xmlns="urn:xmpp:openpgp:0">
 <xs:element name="secretkey" type="xs:base64Binary"/>
 </xs:schema>
-""")
+"""
+)
 DEFAULT_TRUST_MODEL_PARAM = f"""
 <params>
 <individual>
 """Generate a secure passphrase for symmetric encryption.
 @return: The passphrase.
 """
-return "-".join("".join(
+return "-".join(
-secrets.choice("123456789ABCDEFGHIJKLMNPQRSTUVWXYZ") for __ in range(4)
+"".join(secrets.choice("123456789ABCDEFGHIJKLMNPQRSTUVWXYZ") for __ in range(4))
-) for __ in range(6))
+for __ in range(6)
+)
 # TODO: Handle the user id mess
 class XEP_0373:
 """
 xep_0163.add_pep_event(
 "OX_PUBLIC_KEYS_LIST",
 PUBLIC_KEYS_LIST_NODE,
 lambda items_event, profile: defer.ensureDeferred(
 self.__on_public_keys_list_update(items_event, profile)
-)
+),
 )
 async def profile_connecting(self, client):
 client.gpg_provider = get_gpg_provider(self.host, client)
 async def profile_connected(  # pylint: disable=invalid-name
-self,
+self, client: SatXMPPClient
-client: SatXMPPClient
 ) -> None:
 """
 @param client: The client.
 """
 profile = cast(str, client.profile)
 if not profile in self.__storage:
-self.__storage[profile] = \
+self.__storage[profile] = persistent.LazyPersistentBinaryDict(
-persistent.LazyPersistentBinaryDict("XEP-0373", client.profile)
+"XEP-0373", client.profile
+)
 if len(self.list_secret_keys(client)) == 0:
 log.debug(f"Generating first GPG key for {client.jid.userhost()}.")
 await self.create_key(client)
 async def __on_public_keys_list_update(
-self,
+self, items_event: pubsub.ItemsEvent, profile: str
-items_event: pubsub.ItemsEvent,
-profile: str
 ) -> None:
 """Handle public keys list updates fired by PEP.
 @param items_event: The event.
 @param profile: The profile this event belongs to.
 log.debug("Ignoring empty public keys list update.")
 return
 public_keys_list_elt = cast(
 Optional[domish.Element],
-next(item_elt.elements(NS_OX, "public-keys-list"), None)
+next(item_elt.elements(NS_OX, "public-keys-list"), None),
 )
 pubkey_metadata_elts: Optional[List[domish.Element]] = None
 if public_keys_list_elt is not None:
 try:
 PUBLIC_KEYS_LIST_SCHEMA.validate(public_keys_list_elt.toXml())
 except xmlschema.XMLSchemaValidationError:
 pass
 else:
-pubkey_metadata_elts = \
+pubkey_metadata_elts = list(
-list(public_keys_list_elt.elements(NS_OX, "pubkey-metadata"))
+public_keys_list_elt.elements(NS_OX, "pubkey-metadata")
+)
 if pubkey_metadata_elts is None:
 log.warning(f"Malformed public keys list update item: {item_elt.toXml()}")
 return
-new_public_keys_metadata = { PublicKeyMetadata(
+new_public_keys_metadata = {
-fingerprint=cast(str, pubkey_metadata_elt["v4-fingerprint"]),
+PublicKeyMetadata(
-timestamp=parse_datetime(cast(str, pubkey_metadata_elt["date"]))
+fingerprint=cast(str, pubkey_metadata_elt["v4-fingerprint"]),
-) for pubkey_metadata_elt in pubkey_metadata_elts }
+timestamp=parse_datetime(cast(str, pubkey_metadata_elt["date"])),
+)
+for pubkey_metadata_elt in pubkey_metadata_elts
+}
 storage_key = STR_KEY_PUBLIC_KEYS_METADATA.format(sender.userhost())
 local_public_keys_metadata = {
 PublicKeyMetadata.from_dict(pkm)
 # Check whether this update was for our account and make sure all of our keys are
 # included in the update
 if sender.userhost() == client.jid.userhost():
 secret_keys = self.list_secret_keys(client)
-missing_keys = set(filter(lambda secret_key: all(
+missing_keys = set(
-key_metadata.fingerprint != secret_key.public_key.fingerprint
+filter(
-for key_metadata
+lambda secret_key: all(
-in new_public_keys_metadata
+key_metadata.fingerprint != secret_key.public_key.fingerprint
-), secret_keys))
+for key_metadata in new_public_keys_metadata
+),
+secret_keys,
+)
+)
 if len(missing_keys) > 0:
 log.warning(
 "Public keys list update did not contain at least one of our keys."
 f" {new_public_keys_metadata}"
 )
 for missing_key in missing_keys:
 log.warning(missing_key.public_key.fingerprint)
-new_public_keys_metadata.add(PublicKeyMetadata(
+new_public_keys_metadata.add(
-fingerprint=missing_key.public_key.fingerprint,
+PublicKeyMetadata(
-timestamp=datetime.now(timezone.utc)
+fingerprint=missing_key.public_key.fingerprint,
-))
+timestamp=datetime.now(timezone.utc),
+)
+)
 await self.publish_public_keys_list(client, new_public_keys_metadata)
 await self.__storage[profile].force(
-storage_key,
+storage_key, [pkm.to_dict() for pkm in new_public_keys_metadata]
-[pkm.to_dict() for pkm in new_public_keys_metadata]
 )
 def list_public_keys(self, client: SatXMPPClient, jid: jid.JID) -> Set[GPGPublicKey]:
 """List GPG public keys available for a JID.
 public_keys_list = {
 PublicKeyMetadata.from_dict(pkm)
 for pkm in await self.__storage[client.profile].get(storage_key, [])
 }
-public_keys_list.add(PublicKeyMetadata(
+public_keys_list.add(
-fingerprint=secret_key.public_key.fingerprint,
+PublicKeyMetadata(
-timestamp=datetime.now(timezone.utc)
+fingerprint=secret_key.public_key.fingerprint,
-))
+timestamp=datetime.now(timezone.utc),
+)
+)
 await self.publish_public_keys_list(client, public_keys_list)
 await self.__storage[client.profile].force(
-storage_key,
+storage_key, [pkm.to_dict() for pkm in public_keys_list]
-[pkm.to_dict() for pkm in public_keys_list]
 )
 return secret_key
 @staticmethod
 def __build_content_element(
 element_name: Literal["signcrypt", "sign", "crypt"],
 recipient_jids: Iterable[jid.JID],
-include_rpad: bool
+include_rpad: bool,
 ) -> Tuple[domish.Element, domish.Element]:
 """Build a content element.
 @param element_name: The name of the content element.
 @param recipient_jids: The intended recipients of this content element. Can be
 # XEP-0373 doesn't specify bounds for the length of the random padding. This
 # uses the bounds specified in XEP-0420 for the closely related rpad affix.
 rpad_length = secrets.randbelow(201)
 rpad_content = "".join(
 secrets.choice(string.digits + string.ascii_letters + string.punctuation)
-for __
+for __ in range(rpad_length)
-in range(rpad_length)
 )
 content_elt.addElement("rpad", content=rpad_content)
 payload_elt = content_elt.addElement("payload")
 return content_elt, payload_elt
 @staticmethod
 def build_signcrypt_element(
-recipient_jids: Iterable[jid.JID]
+recipient_jids: Iterable[jid.JID],
 ) -> Tuple[domish.Element, domish.Element]:
 """Build a ``<signcrypt/>`` content element.
 @param recipient_jids: The intended recipients of this content element. Can be
 bare JIDs.
 return XEP_0373.__build_content_element("signcrypt", recipient_jids, True)
 @staticmethod
 def build_sign_element(
-recipient_jids: Iterable[jid.JID],
+recipient_jids: Iterable[jid.JID], include_rpad: bool
-include_rpad: bool
 ) -> Tuple[domish.Element, domish.Element]:
 """Build a ``<sign/>`` content element.
 @param recipient_jids: The intended recipients of this content element. Can be
 bare JIDs.
 return XEP_0373.__build_content_element("sign", recipient_jids, include_rpad)
 @staticmethod
 def build_crypt_element(
-recipient_jids: Iterable[jid.JID]
+recipient_jids: Iterable[jid.JID],
 ) -> Tuple[domish.Element, domish.Element]:
 """Build a ``<crypt/>`` content element.
 @param recipient_jids: The intended recipients of this content element. Specifying
 the intended recipients is OPTIONAL for the ``<crypt/>`` content element. Can
 async def build_openpgp_element(
 self,
 client: SatXMPPClient,
 content_elt: domish.Element,
-recipient_jids: Set[jid.JID]
+recipient_jids: Set[jid.JID],
 ) -> domish.Element:
 """Build an ``<openpgp/>`` element.
 @param client: The client to perform this operation with.
 @param content_elt: The content element to contain in the ``<openpgp/>`` element.
 gpg_provider = get_gpg_provider(self.host, client)
 # TODO: I'm not sure whether we want to sign with all keys by default or choose
 # just one key/a subset of keys to sign with.
-signing_keys = set(filter(
+signing_keys = set(
-lambda secret_key: gpg_provider.can_sign(secret_key.public_key),
+filter(
-self.list_secret_keys(client)
+lambda secret_key: gpg_provider.can_sign(secret_key.public_key),
-))
+self.list_secret_keys(client),
+)
+)
 encryption_keys: Set[GPGPublicKey] = set()
 for recipient_jid in recipient_jids:
 # import all keys of the recipient
 async def unpack_openpgp_element(
 self,
 client: SatXMPPClient,
 openpgp_elt: domish.Element,
 element_name: Literal["signcrypt", "sign", "crypt"],
-sender_jid: jid.JID
+sender_jid: jid.JID,
 ) -> Tuple[domish.Element, datetime]:
 """Verify, decrypt and unpack an ``<openpgp/>`` element.
 @param client: The client to perform this operation with.
 @param openpgp_elt: The ``<openpgp/>`` element.
 the calling code.
 """
 gpg_provider = get_gpg_provider(self.host, client)
-decryption_keys = set(filter(
+decryption_keys = set(
-lambda secret_key: gpg_provider.can_encrypt(secret_key.public_key),
+filter(
-self.list_secret_keys(client)
+lambda secret_key: gpg_provider.can_encrypt(secret_key.public_key),
-))
+self.list_secret_keys(client),
+)
+)
 # import all keys of the sender
 all_public_keys = await self.import_all_public_keys(client, sender_jid)
 # Filter for keys that can sign
 openpgp_message = base64.b64decode(str(openpgp_elt))
 content: bytes
 if element_name == "signcrypt":
 content = gpg_provider.decrypt(
-openpgp_message,
+openpgp_message, decryption_keys, public_keys=verification_keys
-decryption_keys,
-public_keys=verification_keys
 )
 elif element_name == "sign":
 content = gpg_provider.verify(openpgp_message, verification_keys)
 elif element_name == "crypt":
 content = gpg_provider.decrypt(openpgp_message, decryption_keys)
 else:
 assert_never(element_name)
 try:
 content_elt = cast(
-domish.Element,
+domish.Element, xml_tools.ElementParser()(content.decode("utf-8"))
-xml_tools.ElementParser()(content.decode("utf-8"))
 )
 except UnicodeDecodeError as e:
 raise exceptions.ParsingError("UTF-8 decoding error") from e
 try:
 ) from e
 if content_elt.name != element_name:
 raise exceptions.ParsingError(f"Not a <{element_name}/> element.")
-recipient_jids = \
+recipient_jids = {
-{ jid.JID(to_elt["jid"]) for to_elt in content_elt.elements(NS_OX, "to") }
+jid.JID(to_elt["jid"]) for to_elt in content_elt.elements(NS_OX, "to")
+}
 if (
-client.jid.userhostJID() not in { jid.userhostJID() for jid in recipient_jids }
+client.jid.userhostJID() not in {jid.userhostJID() for jid in recipient_jids}
 and element_name != "crypt"
 ):
 raise VerificationError(
 f"Recipient list in <{element_name}/> element does not list our (bare)"
 f" JID."
 payload_elt = next(content_elt.elements(NS_OX, "payload"))
 return payload_elt, timestamp
 async def publish_public_key(
-self,
+self, client: SatXMPPClient, public_key: GPGPublicKey
-client: SatXMPPClient,
-public_key: GPGPublicKey
 ) -> None:
 """Publish a public key.
 @param client: The client.
 @param public_key: The public key to publish.
 format_datetime(),
 extra={
 XEP_0060.EXTRA_PUBLISH_OPTIONS: {
 XEP_0060.OPT_PERSIST_ITEMS: "true",
 XEP_0060.OPT_ACCESS_MODEL: "open",
-XEP_0060.OPT_MAX_ITEMS: 1
+XEP_0060.OPT_MAX_ITEMS: 1,
 },
 # TODO: Do we really want publish_without_options here?
-XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options"
+XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options",
-}
+},
 )
 except Exception as e:
 raise XMPPInteractionFailed("Publishing the public key failed.") from e
 async def import_all_public_keys(
-self,
+self, client: SatXMPPClient, entity_jid: jid.JID
-client: SatXMPPClient,
-entity_jid: jid.JID
 ) -> Set[GPGPublicKey]:
 """import all public keys of a JID that have not been imported before.
 @param client: The client.
 @param jid: The JID. Can be a bare JID.
 if not public_keys_metadata:
 public_keys_metadata = await self.download_public_keys_list(
 client, entity_jid
 )
 if not public_keys_metadata:
-raise exceptions.NotFound(
+raise exceptions.NotFound(f"Can't find public keys for {entity_jid}")
-f"Can't find public keys for {entity_jid}"
-)
 else:
 await self.__storage[client.profile].aset(
-storage_key,
+storage_key, [pkm.to_dict() for pkm in public_keys_metadata]
-[pkm.to_dict() for pkm in public_keys_metadata]
 )
+missing_keys = set(
-missing_keys = set(filter(lambda public_key_metadata: all(
+filter(
-public_key_metadata.fingerprint != public_key.fingerprint
+lambda public_key_metadata: all(
-for public_key
+public_key_metadata.fingerprint != public_key.fingerprint
-in available_public_keys
+for public_key in available_public_keys
-), public_keys_metadata))
+),
+public_keys_metadata,
+)
+)
 for missing_key in missing_keys:
 try:
 available_public_keys.add(
-await self.import_public_key(client, entity_jid, missing_key.fingerprint)
+await self.import_public_key(
+client, entity_jid, missing_key.fingerprint
+)
 )
 except Exception as e:
 log.warning(
 f"import of public key {missing_key.fingerprint} owned by"
 f" {entity_jid.userhost()} failed, ignoring: {e}"
 )
 return available_public_keys
 async def import_public_key(
-self,
+self, client: SatXMPPClient, jid: jid.JID, fingerprint: str
-client: SatXMPPClient,
-jid: jid.JID,
-fingerprint: str
 ) -> GPGPublicKey:
 """import a public key.
 @param client: The client.
 @param jid: The JID owning the public key. Can be a bare JID.
 node = f"urn:xmpp:openpgp:0:public-keys:{fingerprint}"
 try:
 items, __ = await self.__xep_0060.get_items(
-client,
+client, jid.userhostJID(), node, max_items=1
-jid.userhostJID(),
-node,
-max_items=1
 )
 except exceptions.NotFound as e:
 raise exceptions.NotFound(
 f"No public key with fingerprint {fingerprint} published by JID"
 f" {jid.userhost()}."
 f"No public key with fingerprint {fingerprint} published by JID"
 f" {jid.userhost()}."
 ) from e
 pubkey_elt = cast(
-Optional[domish.Element],
+Optional[domish.Element], next(item_elt.elements(NS_OX, "pubkey"), None)
-next(item_elt.elements(NS_OX, "pubkey"), None)
 )
 if pubkey_elt is None:
 raise exceptions.ParsingError(
 f"Publish-Subscribe item of JID {jid.userhost()} doesn't contain pubkey"
 raise exceptions.ParsingError(
 f"Publish-Subscribe item of JID {jid.userhost()} doesn't pass pubkey"
 f" schema validation."
 ) from e
-public_key = gpg_provider.import_public_key(base64.b64decode(str(
+public_key = gpg_provider.import_public_key(
-next(pubkey_elt.elements(NS_OX, "data"))
+base64.b64decode(str(next(pubkey_elt.elements(NS_OX, "data"))))
-)))
+)
 return public_key
 async def publish_public_keys_list(
-self,
+self, client: SatXMPPClient, public_keys_list: Iterable[PublicKeyMetadata]
-client: SatXMPPClient,
-public_keys_list: Iterable[PublicKeyMetadata]
 ) -> None:
 """Publish/update the own public keys list.
 @param client: The client.
 @param public_keys_list: The public keys list.
 @warning: All public keys referenced in the public keys list MUST be published
 beforehand.
 """
-if len({ pkm.fingerprint for pkm in public_keys_list }) != len(public_keys_list):
+if len({pkm.fingerprint for pkm in public_keys_list}) != len(public_keys_list):
 raise ValueError("Public keys list contains duplicate fingerprints.")
 node = "urn:xmpp:openpgp:0:public-keys"
 public_keys_list_elt = domish.Element((NS_OX, "public-keys-list"))
 item_id=XEP_0060.ID_SINGLETON,
 extra={
 XEP_0060.EXTRA_PUBLISH_OPTIONS: {
 XEP_0060.OPT_PERSIST_ITEMS: "true",
 XEP_0060.OPT_ACCESS_MODEL: "open",
-XEP_0060.OPT_MAX_ITEMS: 1
+XEP_0060.OPT_MAX_ITEMS: 1,
 },
 # TODO: Do we really want publish_without_options here?
-XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options"
+XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options",
-}
+},
 )
 except Exception as e:
 raise XMPPInteractionFailed("Publishing the public keys list failed.") from e
 async def download_public_keys_list(
-self,
+self, client: SatXMPPClient, jid: jid.JID
-client: SatXMPPClient,
-jid: jid.JID
 ) -> Optional[Set[PublicKeyMetadata]]:
 """Download the public keys list of a JID.
 @param client: The client.
 @param jid: The JID. Can be a bare JID.
 node = "urn:xmpp:openpgp:0:public-keys"
 try:
 items, __ = await self.__xep_0060.get_items(
-client,
+client, jid.userhostJID(), node, max_items=1
-jid.userhostJID(),
-node,
-max_items=1
 )
 except exceptions.NotFound:
 return None
 except Exception as e:
 raise XMPPInteractionFailed() from e
 except IndexError:
 return None
 public_keys_list_elt = cast(
 Optional[domish.Element],
-next(item_elt.elements(NS_OX, "public-keys-list"), None)
+next(item_elt.elements(NS_OX, "public-keys-list"), None),
 )
 if public_keys_list_elt is None:
 return None
 ) from e
 return {
 PublicKeyMetadata(
 fingerprint=pubkey_metadata_elt["v4-fingerprint"],
-timestamp=parse_datetime(pubkey_metadata_elt["date"])
+timestamp=parse_datetime(pubkey_metadata_elt["date"]),
 )
-for pubkey_metadata_elt
+for pubkey_metadata_elt in public_keys_list_elt.elements(
-in public_keys_list_elt.elements(NS_OX, "pubkey-metadata")
+NS_OX, "pubkey-metadata"
+)
 }
 async def __prepare_secret_key_synchronization(
-self,
+self, client: SatXMPPClient
-client: SatXMPPClient
 ) -> Optional[domish.Element]:
 """Prepare for secret key synchronization.
 Makes sure the relative protocols and protocol extensions are supported by the
 server and makes sure that the PEP node for secret synchronization exists and is
 protocols or protocol extensions.
 @raise XMPPInteractionFailed: if any interaction via XMPP failed.
 """
 try:
-infos = cast(DiscoInfo, await self.host.memory.disco.get_infos(
+infos = cast(
-client,
+DiscoInfo,
-client.jid.userhostJID()
+await self.host.memory.disco.get_infos(client, client.jid.userhostJID()),
-))
+)
 except Exception as e:
 raise XMPPInteractionFailed(
 "Error performing service discovery on the own bare JID."
 ) from e
 if "http://jabber.org/protocol/pubsub#access-whitelist" not in features:
 raise exceptions.FeatureNotFound(
 "Server doesn't support the whitelist access model."
 )
-persistent_items_supported = \
+persistent_items_supported = (
 "http://jabber.org/protocol/pubsub#persistent-items" in features
+)
 # TODO: persistent-items is a SHOULD, how do we handle the feature missing?
 node = "urn:xmpp:openpgp:0:secret-key"
 try:
 items, __ = await self.__xep_0060.get_items(
-client,
+client, client.jid.userhostJID(), node, max_items=1
-client.jid.userhostJID(),
-node,
-max_items=1
 )
 except exceptions.NotFound:
 try:
 await self.__xep_0060.createNode(
 client,
 client.jid.userhostJID(),
 node,
 {
 XEP_0060.OPT_PERSIST_ITEMS: "true",
 XEP_0060.OPT_ACCESS_MODEL: "whitelist",
-XEP_0060.OPT_MAX_ITEMS: "1"
+XEP_0060.OPT_MAX_ITEMS: "1",
-}
+},
 )
 except Exception as e:
 raise XMPPInteractionFailed(
 "Error creating the secret key synchronization node."
 ) from e
 return cast(domish.Element, items[0])
 except IndexError:
 return None
 async def export_secret_keys(
-self,
+self, client: SatXMPPClient, secret_keys: Iterable[GPGSecretKey]
-client: SatXMPPClient,
-secret_keys: Iterable[GPGSecretKey]
 ) -> str:
 """Export secret keys to synchronize them with other devices.
 @param client: The client.
 @param secret_keys: The secret keys to export.
 secretkey_elt = domish.Element((NS_OX, "secretkey"))
 secretkey_elt.addContent(base64.b64encode(ciphertext).decode("ASCII"))
 try:
 await self.__xep_0060.send_item(
-client,
+client, client.jid.userhostJID(), node, secretkey_elt
-client.jid.userhostJID(),
-node,
-secretkey_elt
 )
 except Exception as e:
 raise XMPPInteractionFailed("Publishing the secret keys failed.") from e
 return backup_code
 item_elt = await self.__prepare_secret_key_synchronization(client)
 if item_elt is None:
 return None
 secretkey_elt = cast(
-Optional[domish.Element],
+Optional[domish.Element], next(item_elt.elements(NS_OX, "secretkey"), None)
-next(item_elt.elements(NS_OX, "secretkey"), None)
 )
 if secretkey_elt is None:
 return None
 ) from e
 return base64.b64decode(str(secretkey_elt))
 def import_secret_keys(
-self,
+self, client: SatXMPPClient, ciphertext: bytes, backup_code: str
-client: SatXMPPClient,
-ciphertext: bytes,
-backup_code: str
 ) -> Set[GPGSecretKey]:
 """import previously downloaded secret keys.
 The downloading and importing steps are separate since a backup code is required
 for the import and it should be possible to try multiple backup codes without
 @raise DecryptionFailed: on decryption failure.
 """
 gpg_provider = get_gpg_provider(self.host, client)
-return gpg_provider.restore_secret_keys(gpg_provider.decrypt_symmetrically(
+return gpg_provider.restore_secret_keys(
-ciphertext,
+gpg_provider.decrypt_symmetrically(ciphertext, backup_code)
-backup_code
+)
-))
 @staticmethod
 def __get_joined_muc_users(
-client: SatXMPPClient,
+client: SatXMPPClient, xep_0045: XEP_0045, room_jid: jid.JID
-xep_0045: XEP_0045,
-room_jid: jid.JID
 ) -> Set[jid.JID]:
 """
 @param client: The client.
 @param xep_0045: A MUC plugin instance.
 @param room_jid: The room JID.
 bare_jids.add(entity.jid.userhostJID())
 return bare_jids
 async def get_trust(
-self,
+self, client: SatXMPPClient, public_key: GPGPublicKey, owner: jid.JID
-client: SatXMPPClient,
-public_key: GPGPublicKey,
-owner: jid.JID
 ) -> TrustLevel:
 """Query the trust level of a public key.
 @param client: The client to perform this operation under.
 @param public_key: The public key.
 async def set_trust(
 self,
 client: SatXMPPClient,
 public_key: GPGPublicKey,
 owner: jid.JID,
-trust_level: TrustLevel
+trust_level: TrustLevel,
 ) -> None:
 """Set the trust level of a public key.
 @param client: The client to perform this operation under.
 @param public_key: The public key.
 key = f"/trust/{owner.userhost()}/{public_key.fingerprint}"
 await self.__storage[client.profile].force(key, trust_level.name)
 async def get_trust_ui(  # pylint: disable=invalid-name
-self,
+self, client: SatXMPPClient, entity: jid.JID
-client: SatXMPPClient,
-entity: jid.JID
 ) -> xml_tools.XMLUI:
 """
 @param client: The client.
 @param entity: The entity whose device trust levels to manage.
 @return: An XMLUI instance which opens a form to manage the trust level of all
 bare_jids: Set[jid.JID]
 if self.__xep_0045 is not None and self.__xep_0045.is_joined_room(client, entity):
 bare_jids = self.__get_joined_muc_users(client, self.__xep_0045, entity)
 else:
-bare_jids = { entity.userhostJID() }
+bare_jids = {entity.userhostJID()}
-all_public_keys = list({
+all_public_keys = list(
-bare_jid: list(self.list_public_keys(client, bare_jid))
+{
-for bare_jid
+bare_jid: list(self.list_public_keys(client, bare_jid))
-in bare_jids
+for bare_jid in bare_jids
-}.items())
+}.items()
+)
 async def callback(
-data: Any,
+data: Any, profile: str  # pylint: disable=unused-argument
-profile: str  # pylint: disable=unused-argument
 ) -> Dict[Never, Never]:
 """
 @param data: The XMLUI result produces by the trust UI form.
 @param profile: The profile.
 @return: An empty dictionary. The type of the return value was chosen
 if C.bool(data.get("cancelled", "false")):
 return {}
 data_form_result = cast(
-Dict[str, str],
+Dict[str, str], xml_tools.xmlui_result_2_data_form_result(data)
-xml_tools.xmlui_result_2_data_form_result(data)
 )
 for key, value in data_form_result.items():
 if not key.startswith("trust_"):
 continue
 return {}
 submit_id = self.host.register_callback(callback, with_data=True, one_shot=True)
 result = xml_tools.XMLUI(
-panel_type=C.XMLUI_FORM,
+panel_type=C.XMLUI_FORM, title=D_("OX trust management"), submit_id=submit_id
-title=D_("OX trust management"),
-submit_id=submit_id
 )
 # Casting this to Any, otherwise all calls on the variable cause type errors
 # pylint: disable=no-member
 trust_ui = cast(Any, result)
-trust_ui.addText(D_(
+trust_ui.addText(
-"This is OX trusting system. You'll see below the GPG keys of your "
+D_(
-"contacts, and a list selection to trust them or not. A trusted key "
+"This is OX trusting system. You'll see below the GPG keys of your "
-"can read your messages in plain text, so be sure to only validate "
+"contacts, and a list selection to trust them or not. A trusted key "
-"keys that you are sure are belonging to your contact. It's better "
+"can read your messages in plain text, so be sure to only validate "
-"to do this when you are next to your contact, so "
+"keys that you are sure are belonging to your contact. It's better "
-"you can check the \"fingerprint\" of the key "
+"to do this when you are next to your contact, so "
-"yourself. Do *not* validate a key if the fingerprint is wrong!"
+'you can check the "fingerprint" of the key '
-))
+"yourself. Do *not* validate a key if the fingerprint is wrong!"
+)
+)
 own_secret_keys = self.list_secret_keys(client)
 trust_ui.change_container("label")
 for index, secret_key in enumerate(own_secret_keys):
 trust_ui.addLabel(D_(f"Own secret key {index} fingerprint"))
 trust_ui.addText(secret_key.public_key.fingerprint)
 trust_ui.addEmpty()
 trust_ui.addEmpty()
-for outer_index, [ owner, public_keys ] in enumerate(all_public_keys):
+for outer_index, [owner, public_keys] in enumerate(all_public_keys):
 for inner_index, public_key in enumerate(public_keys):
 trust_ui.addLabel(D_("Contact"))
 trust_ui.addJid(jid.JID(owner))
 trust_ui.addLabel(D_("Fingerprint"))
 trust_ui.addText(public_key.fingerprint)
 trust_ui.addLabel(D_("Trust this device?"))
 current_trust_level = await self.get_trust(client, public_key, owner)
-avaiable_trust_levels = \
+avaiable_trust_levels = {
-{ TrustLevel.DISTRUSTED, TrustLevel.TRUSTED, current_trust_level }
+TrustLevel.DISTRUSTED,
+TrustLevel.TRUSTED,
+current_trust_level,
+}
 trust_ui.addList(
 f"trust_{outer_index}_{inner_index}",
-options=[ trust_level.name for trust_level in avaiable_trust_levels ],
+options=[trust_level.name for trust_level in avaiable_trust_levels],
 selected=current_trust_level.name,
-styles=[ "inline" ]
+styles=["inline"],
 )
 trust_ui.addEmpty()
 trust_ui.addEmpty()

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0373.py @ 4270:0d7bb4df2343