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.
This commit is contained in:
Goffi
parent
7b2afd238b
commit
14a50673a1
1 changed files with 58 additions and 27 deletions
|
|
@ -16,12 +16,12 @@
|
|||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
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("<hash> element not found")
|
||||
algo = hash_elt.getAttribute("algo")
|
||||
if algo is None:
|
||||
raise exceptions.DataError('Missing "algo" attribute on <hash> 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 <hash-used> 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 <hash> 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: <hash> 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
|
||||
|
|
|
|||
Loading…
Reference in a new issue