libervia-backend: libervia/backend/plugins/plugin_xep

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

Reformatted code base using black.

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

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 "ProfileRequirementsNotMet",
 "AffixVerificationFailed",
 "SCECustomAffix",
 "SCEAffixPolicy",
 "SCEProfile",
-"SCEAffixValues"
+"SCEAffixValues",
 ]
 log = cast(Logger, getLogger(__name__))  # type: ignore[no-untyped-call]
 PLUGIN_INFO = {
 C.PI_NAME: "SCE",
 C.PI_IMPORT_NAME: "XEP-0420",
 C.PI_TYPE: "SEC",
-C.PI_PROTOCOLS: [ "XEP-0420" ],
+C.PI_PROTOCOLS: ["XEP-0420"],
-C.PI_DEPENDENCIES: [ "XEP-0334", "XEP-0082" ],
+C.PI_DEPENDENCIES: ["XEP-0334", "XEP-0082"],
-C.PI_RECOMMENDATIONS: [ "XEP-0045", "XEP-0033", "XEP-0359" ],
+C.PI_RECOMMENDATIONS: ["XEP-0045", "XEP-0033", "XEP-0359"],
 C.PI_MAIN: "XEP_0420",
 C.PI_HANDLER: "no",
 C.PI_DESCRIPTION: D_("Implementation of Stanza Content Encryption"),
 }
 NS_HINTS,
 NS_SID,  # TODO: Not sure whether this ban applies to both stanza-id and origin-id
 NS_ADDRESS,
 # Not part of the specification (yet), but just doesn't make sense in an encrypted
 # envelope:
-NS_EME
+NS_EME,
 }
 # Set of (namespace, element name) tuples that define elements which are never allowed
 # to be transferred in an encrypted envelope. If all elements under a certain
 # namespace are forbidden, the namespace can be added to
 content_byte_size = len(content.toXml().encode("utf-8"))
 content_byte_size_diff = content_byte_size - empty_content_byte_size
 rpad_length = max(0, 53 - content_byte_size_diff) + 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)
 )
 envelope.addElement((NS_SCE, "rpad"), content=rpad_content)
 if profile.time_policy is not SCEAffixPolicy.NOT_NEEDED:
 time_element = envelope.addElement((NS_SCE, "time"))
 return envelope.toXml().encode("utf-8")
 @staticmethod
 def unpack_stanza(
-profile: SCEProfile,
+profile: SCEProfile, stanza: domish.Element, envelope_serialized: bytes
-stanza: domish.Element,
-envelope_serialized: bytes
 ) -> SCEAffixValues:
 """Unpack a stanza packed according to Stanza Content Encryption.
 Parses the serialized envelope as XML, verifies included affixes and makes sure
 the requirements of the profile are met, and restores the stanza by moving
 ) from e
 custom_affixes = set(profile.custom_policies.keys())
 # Make sure the envelope adheres to the schema
-parser = etree.XMLParser(schema=etree.XMLSchema(etree.XML(ENVELOPE_SCHEMA.format(
+parser = etree.XMLParser(
-custom_affix_references="".join(
+schema=etree.XMLSchema(
-f'<xs:element ref="{custom_affix.element_name}" minOccurs="0"/>'
+etree.XML(
-for custom_affix
+ENVELOPE_SCHEMA.format(
-in custom_affixes
+custom_affix_references="".join(
-),
+f'<xs:element ref="{custom_affix.element_name}" minOccurs="0"/>'
-custom_affix_definitions="".join(
+for custom_affix in custom_affixes
-custom_affix.element_schema
+),
-for custom_affix
+custom_affix_definitions="".join(
-in custom_affixes
+custom_affix.element_schema for custom_affix in custom_affixes
+),
+).encode("utf-8")
+)
 )
-).encode("utf-8"))))
+)
 try:
 etree.fromstring(envelope_serialized_string, parser)
 except etree.XMLSyntaxError as e:
 raise exceptions.ParsingError(
 envelope = cast(domish.Element, ElementParser()(envelope_serialized_string))
 content = next(envelope.elements(NS_SCE, "content"))
 # Verify the affixes
 rpad_element = cast(
-Optional[domish.Element],
+Optional[domish.Element], next(envelope.elements(NS_SCE, "rpad"), None)
-next(envelope.elements(NS_SCE, "rpad"), None)
 )
 time_element = cast(
-Optional[domish.Element],
+Optional[domish.Element], next(envelope.elements(NS_SCE, "time"), None)
-next(envelope.elements(NS_SCE, "time"), None)
 )
 to_element = cast(
-Optional[domish.Element],
+Optional[domish.Element], next(envelope.elements(NS_SCE, "to"), None)
-next(envelope.elements(NS_SCE, "to"), None)
 )
 from_element = cast(
-Optional[domish.Element],
+Optional[domish.Element], next(envelope.elements(NS_SCE, "from"), None)
-next(envelope.elements(NS_SCE, "from"), None)
 )
 # The rpad doesn't need verification.
 rpad_value = None if rpad_element is None else str(rpad_element)
 # The time affix isn't verified other than that the timestamp is parseable.
 try:
-timestamp_value = None if time_element is None else \
+timestamp_value = (
-XEP_0082.parse_datetime(time_element["stamp"])
+None
+if time_element is None
+else XEP_0082.parse_datetime(time_element["stamp"])
+)
 except ValueError as e:
 raise AffixVerificationFailed("Malformed time affix.") from e
 # The to affix is verified by comparing the to attribute of the stanza with the
 # JID referenced by the affix. Note that only bare JIDs are compared as per the
 custom_values: Dict[SCECustomAffix, domish.Element] = {}
 for affix in custom_affixes:
 element_name = affix.element_name
 element = cast(
 Optional[domish.Element],
-next(envelope.elements(NS_SCE, element_name), None)
+next(envelope.elements(NS_SCE, element_name), None),
 )
 if element is not None:
 affix.verify(stanza, element)
 custom_values[affix] = element
 # Check whether all affixes required by the profile are present
-rpad_missing = \
+rpad_missing = (
 profile.rpad_policy is SCEAffixPolicy.REQUIRED and rpad_element is None
-time_missing = \
+)
+time_missing = (
 profile.time_policy is SCEAffixPolicy.REQUIRED and time_element is None
-to_missing = \
+)
-profile.to_policy is SCEAffixPolicy.REQUIRED and to_element is None
+to_missing = profile.to_policy is SCEAffixPolicy.REQUIRED and to_element is None
-from_missing = \
+from_missing = (
 profile.from_policy is SCEAffixPolicy.REQUIRED and from_element is None
+)
 custom_missing = any(
 affix not in custom_values
-for affix, policy
+for affix, policy in profile.custom_policies.items()
-in profile.custom_policies.items()
 if policy is SCEAffixPolicy.REQUIRED
 )
 if rpad_missing or time_missing or to_missing or from_missing or custom_missing:
 custom_missing_string = ""
 # Add the child to the stanza
 stanza.addChild(child)
 return SCEAffixValues(
-rpad_value,
+rpad_value, timestamp_value, recipient_value, sender_value, custom_values
-timestamp_value,
+)
-recipient_value,
-sender_value,
-custom_values
-)

Mercurial > libervia-backend

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