From 14a50673a1e6356305b924861b356235c022c656 Mon Sep 17 00:00:00 2001 From: Goffi Date: Tue, 7 Jul 2026 16:32:00 +0200 Subject: [PATCH] plugin XEP-0300: Fix hash encoding and use `StrEnum` for hashes: - Fix the wrong encoding of the hash hex string value instead of the binary value. - Use a `StrEnum` for model, use models specified in XEP-0414. - Minor typing hint/docstring improvments. --- libervia/backend/plugins/plugin_xep_0300.py | 85 ++++++++++++++------- 1 file changed, 58 insertions(+), 27 deletions(-) diff --git a/libervia/backend/plugins/plugin_xep_0300.py b/libervia/backend/plugins/plugin_xep_0300.py index eeb782a3..f7a571b6 100644 --- a/libervia/backend/plugins/plugin_xep_0300.py +++ b/libervia/backend/plugins/plugin_xep_0300.py @@ -16,12 +16,12 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from _hashlib import HASH import base64 -from collections import OrderedDict +from enum import StrEnum import hashlib from typing import BinaryIO, Callable, Self, TYPE_CHECKING -from _hashlib import HASH from pydantic import BaseModel, Field from twisted.internet import threads from twisted.internet import defer @@ -37,6 +37,9 @@ from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger +Hasher = HASH | hashlib.blake2b + + if TYPE_CHECKING: from libervia.backend.core.main import LiberviaBackend @@ -48,7 +51,7 @@ PLUGIN_INFO = { C.PI_IMPORT_NAME: "XEP-0300", C.PI_TYPE: "XEP", C.PI_MODES: C.PLUG_MODE_BOTH, - C.PI_PROTOCOLS: ["XEP-0300"], + C.PI_PROTOCOLS: ["XEP-0300", "XEP-0414"], C.PI_MAIN: "XEP_0300", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("""Management of cryptographic hashes"""), @@ -57,7 +60,32 @@ PLUGIN_INFO = { NS_HASHES = "urn:xmpp:hashes:2" NS_HASHES_FUNCTIONS = "urn:xmpp:hash-function-text-names:{}" BUFFER_SIZE = 2**12 -ALGO_DEFAULT = "sha-256" + + +class HashAlgo(StrEnum): + """Hash algorithms that can be used, following XEP-0414.""" + + # Kepts for backward compatibility, avoid using SHA-1 + SHA_1 = "sha-1" + SHA_256 = "sha-256" + SHA_512 = "sha-512" + SHA3_256 = "sha3-256" + SHA3_512 = "sha3-512" + BLAKE2B_256 = "blake2b-256" + BLAKE2B_512 = "blake2b-512" + + +ALGOS: dict[HashAlgo, Callable[[], Hasher]] = { + HashAlgo.SHA_1: hashlib.sha1, + HashAlgo.SHA_256: hashlib.sha256, + HashAlgo.SHA_512: hashlib.sha512, + HashAlgo.SHA3_256: hashlib.sha3_256, + HashAlgo.SHA3_512: hashlib.sha3_512, + HashAlgo.BLAKE2B_256: lambda: hashlib.blake2b(digest_size=32), + HashAlgo.BLAKE2B_512: hashlib.blake2b, +} + +ALGO_DEFAULT = HashAlgo.SHA_256 class Hash(BaseModel): @@ -80,6 +108,8 @@ class Hash(BaseModel): if hash_elt.uri != NS_HASHES or hash_elt.name != "hash": raise exceptions.NotFound(" element not found") algo = hash_elt.getAttribute("algo") + if algo is None: + raise exceptions.DataError('Missing "algo" attribute on element') hash_value = str(hash_elt) return cls(algo=algo, hash_=hash_value) @@ -130,6 +160,8 @@ class HashUsed(BaseModel): else: hash_used_elt = child_hash_used_elt algo = hash_used_elt.getAttribute("algo") + if algo is None: + raise exceptions.DataError('Missing "algo" attribute on element') return cls(algo=algo) def to_element(self) -> domish.Element: @@ -143,26 +175,18 @@ class HashUsed(BaseModel): class XEP_0300: - # TODO: add blake after moving to Python 3 - ALGOS: OrderedDict[str, Callable] = OrderedDict( - ( - ("md5", hashlib.md5), - ("sha-1", hashlib.sha1), - ("sha-256", hashlib.sha256), - ("sha-512", hashlib.sha512), - ) - ) + ALGOS = ALGOS ALGO_DEFAULT = ALGO_DEFAULT def __init__(self, host: "LiberviaBackend"): - log.info(_("plugin Hashes initialization")) + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") host.register_namespace("hashes", NS_HASHES) self.host = host def get_handler(self, client: SatXMPPEntity) -> XMPPHandler: return XEP_0300_handler() - def get_hasher(self, algo: str = ALGO_DEFAULT) -> Callable: + def get_hasher(self, algo: str = ALGO_DEFAULT) -> Hasher: """Return hasher instance @param algo: one of the XEP_300.ALGOS keys @@ -170,7 +194,14 @@ class XEP_0300: update method need to be called for each chunk digest or hexdigest can be used at the end """ - return self.ALGOS[algo]() + if algo == HashAlgo.SHA_1: + log.warning( + _( + "SHA-1 is cryptographically broken and should not be used. Consider " + "using SHA-256 or stronger." + ) + ) + return self.ALGOS[HashAlgo(algo)]() def get_default_algo(self) -> str: return ALGO_DEFAULT @@ -196,10 +227,10 @@ class XEP_0300: ) return algo - def _calculate_hash_blocking(self, file_obj: BinaryIO, hasher: HASH) -> str: + def _calculate_hash_blocking(self, file_obj: BinaryIO, hasher: Hasher) -> str: """Calculate hash in a blocking way - /!\\ blocking method, please use calculate_hash instead + ⚠️ blocking method, please use calculate_hash instead @param file_obj: a file-like object @param hasher: the method to call to initialise hash object @return: the hex digest of the hash @@ -211,7 +242,7 @@ class XEP_0300: hasher.update(buf) return hasher.hexdigest() - def calculate_hash(self, file_obj: BinaryIO, hasher: HASH) -> defer.Deferred[str]: + def calculate_hash(self, file_obj: BinaryIO, hasher: Hasher) -> defer.Deferred[str]: return threads.deferToThread(self._calculate_hash_blocking, file_obj, hasher) async def calculate_hash_elt( @@ -243,13 +274,13 @@ class XEP_0300: return hash_used_model.algo def build_hash_elt(self, hash_hex: str, algo: str = ALGO_DEFAULT) -> domish.Element: - """Compute hash and build hash element + """Build a element from a hexadecimal hash. - @param hash_: Hexadecimal representation of hash to use. - @param algo: Algorithm to use, must be a key of XEP_0300.ALGOS. + @param hash_hex: Hexadecimal representation of the hash digest. + @param algo: Algorithm name (e.g. "sha-256"). @return: element """ - b64_hash = base64.b64encode(hash_hex.encode()).decode() + b64_hash = base64.b64encode(bytes.fromhex(hash_hex)).decode() hash_model = Hash(algo=algo, hash_=b64_hash) return hash_model.to_element() @@ -263,7 +294,7 @@ class XEP_0300: @raise exceptions.NotFound: the element is not present @raise exceptions.DataError: the element is invalid """ - algos = list(XEP_0300.ALGOS.keys()) + algos: list[HashAlgo] = list(XEP_0300.ALGOS.keys()) hash_elt = None best_algo = None best_value = None @@ -271,15 +302,15 @@ class XEP_0300: hash_model = Hash.from_element(hash_elt) algo = hash_model.algo try: - idx = algos.index(algo) + idx = algos.index(HashAlgo(algo)) except ValueError: log.warning(f"Proposed {algo} algorithm is not managed") algo = None continue - if best_algo is None or algos.index(best_algo) < idx: + if best_algo is None or algos.index(HashAlgo(best_algo)) < idx: best_algo = algo - best_value = base64.b64decode(hash_model.hash_).decode() + best_value = base64.b64decode(hash_model.hash_).hex() if not hash_elt: raise exceptions.NotFound