From 050c474028179a7c0634fb632e91391bc112b88d Mon Sep 17 00:00:00 2001 From: Prateek99-s <145468640+Prateek99-s@users.noreply.github.com> Date: Sat, 31 Oct 2026 06:28:09 +0000 Subject: [PATCH] typing: annotate repoobj module boundaries, enable strict mypy checks, refs #10021 --- pyproject.toml | 14 ++++++++++ src/borg/compress.pyi | 14 +++++----- src/borg/legacy/repoobj.py | 34 +++++++++++++++--------- src/borg/repoobj.py | 53 ++++++++++++++++++++++++-------------- 4 files changed, 77 insertions(+), 38 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c8c82ef8e0..3b2c27871d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -175,13 +175,27 @@ module = [ "msgpack.*", "llfuse", "pyfuse3", + "mfusepy", "trio", + "borghash.*", + "borgstore.*", + "textual.*", + "backports.*", "borg.crypto.low_level", "borg.legacy.crypto.low_level", "borg.platform.*", ] ignore_missing_imports = true +[[tool.mypy.overrides]] +module = [ + "borg.repoobj", + "borg.legacy.repoobj", +] +disallow_untyped_defs = true +disallow_incomplete_defs = true +check_untyped_defs = true + [tool.tox] requires = ["tox>=4.19", "pkgconfig", "cython", "wheel", "setuptools-scm[toml]"] # Important: when adding/removing Python versions here, diff --git a/src/borg/compress.pyi b/src/borg/compress.pyi index 723e3cee13..46cd82ba91 100644 --- a/src/borg/compress.pyi +++ b/src/borg/compress.pyi @@ -10,20 +10,22 @@ def forget_zstd_compressors(keep_alive: bool = ...) -> None: ... class Compressor: def __init__(self, name: Any = ..., **kwargs) -> None: ... - def compress(self, meta: dict, data: bytes) -> tuple[dict, bytes]: ... - def decompress(self, meta: dict, data: bytes) -> tuple[dict, bytes]: ... + def compress(self, meta: dict, data: bytes | memoryview) -> tuple[dict, bytes]: ... + def decompress(self, meta: dict, data: bytes | memoryview) -> tuple[dict, bytes]: ... @staticmethod def detect(data: bytes) -> Any: ... class CompressorBase: ID: int = ... name: str = ... + level: int = ... + legacy_mode: bool = ... @classmethod def detect(self, data: bytes) -> bool: ... def __init__(self, level: int = ..., **kwargs) -> None: ... def decide(self, data: bytes) -> Any: ... - def compress(self, data: bytes) -> bytes: ... - def decompress(self, data: bytes) -> bytes: ... + def compress(self, meta: dict, data: bytes | memoryview) -> tuple[dict, bytes]: ... + def decompress(self, meta: dict, data: bytes | memoryview) -> tuple[dict, bytes]: ... class Auto(CompressorBase): def __init__(self, compressor: Any) -> None: ... @@ -57,7 +59,7 @@ class ZSTD(DecidingCompressor): def __init__(self, level: int = ..., **kwargs) -> None: ... level: int -LZ4_COMPRESSOR: type[LZ4] -NONE_COMPRESSOR: type[CNONE] +LZ4_COMPRESSOR: LZ4 +NONE_COMPRESSOR: CNONE COMPRESSOR_TABLE: dict diff --git a/src/borg/legacy/repoobj.py b/src/borg/legacy/repoobj.py index ab1b5431a6..db7da989c7 100644 --- a/src/borg/legacy/repoobj.py +++ b/src/borg/legacy/repoobj.py @@ -3,6 +3,8 @@ Moved from borg.repoobj as part of the legacy code separation. """ +from typing import Any + from ..constants import * # NOQA from ..helpers import workarounds from ..compress import Compressor, get_compressor @@ -12,6 +14,9 @@ class RepoObj1: # legacy + key: Any + compressor: Any + def set_assert_id_place(self, place: str) -> None: # borg 1.x crypto has no AEAD with the chunk id in the AAD, so the id check is a needed part of # the read path here: parse always does it and there is nothing to configure (see BORG_ASSERT_ID). @@ -22,7 +27,7 @@ def extract_crypted_data(cls, data: bytes) -> bytes: # used for crypto type detection return data - def __init__(self, key): + def __init__(self, key: Any) -> None: self.key = key self.compressor = get_compressor("lz4", legacy_mode=True) @@ -32,13 +37,13 @@ def id_hash(self, data: bytes) -> bytes: def format( self, id: bytes, - meta: dict, - data: bytes, + meta: dict[str, Any], + data: bytes | memoryview, compress: bool = True, - size: int = None, - ctype: int = None, - clevel: int = None, - ro_type: str = None, + size: int | None = None, + ctype: int | None = None, + clevel: int | None = None, + ro_type: str | None = None, ) -> bytes: assert isinstance(id, bytes) assert meta == {} @@ -54,20 +59,25 @@ def format( data_encrypted = self.key.encrypt(id, data_compressed) return data_encrypted - def parse_meta(self, id: bytes, cdata: bytes) -> dict: + def parse_meta(self, id: bytes, cdata: bytes | memoryview) -> dict[str, Any]: raise NotImplementedError("parse_meta is not available for RepoObj1") def parse( - self, id: bytes, cdata: bytes, decompress: bool = True, want_compressed: bool = False, ro_type: str = None - ) -> tuple[dict, bytes]: + self, + id: bytes, + cdata: bytes | memoryview, + decompress: bool = True, + want_compressed: bool = False, + ro_type: str | None = None, + ) -> tuple[dict[str, Any] | None, bytes | None]: assert not (not decompress and not want_compressed), "invalid parameter combination!" assert isinstance(id, bytes) - assert isinstance(cdata, bytes) + assert isinstance(cdata, (bytes, memoryview)) assert ro_type is not None data_compressed = self.key.decrypt(id, cdata) compressor_cls, compression_level = Compressor.detect(data_compressed[:2]) compressor = compressor_cls(level=compression_level, legacy_mode=True) - meta_compressed = {} + meta_compressed: dict[str, Any] = {} meta_compressed["ctype"] = compressor.ID meta_compressed["clevel"] = compressor.level meta_compressed["csize"] = len(data_compressed) diff --git a/src/borg/repoobj.py b/src/borg/repoobj.py index 750000f245..884da2f296 100644 --- a/src/borg/repoobj.py +++ b/src/borg/repoobj.py @@ -1,11 +1,12 @@ import os -from collections import namedtuple +from collections.abc import Callable from struct import Struct +from typing import Any, NamedTuple from .constants import * # NOQA from .helpers import msgpack, workarounds from .helpers.errors import Error, IntegrityError -from .compress import Compressor, LZ4_COMPRESSOR +from .compress import Compressor, CompressorBase, LZ4_COMPRESSOR from .crypto.low_level import IntegrityError as IntegrityErrorBase # Workaround for lost passphrase or key in the "authenticated-*" modes @@ -31,7 +32,7 @@ ASSERT_ID_PLACE_DEFAULT = "read" -def get_assert_id_places(): +def get_assert_id_places() -> frozenset[str]: """Determine the configurable places that shall verify the chunk id, see the BORG_ASSERT_ID docs.""" value = os.environ.get("BORG_ASSERT_ID") if value is None: @@ -75,9 +76,20 @@ def get_assert_id_places(): class RepoObj: + class ObjHeader(NamedTuple): + magic: bytes + version: int + chunk_id: bytes + meta_size: int + data_size: int + # Object header: magic (8b), format version (1b), chunk_id (32b), meta size (4b), data size (4b). - obj_header = Struct("<8sB32sII") - ObjHeader = namedtuple("ObjHeader", "magic version chunk_id meta_size data_size") + obj_header: Struct = Struct("<8sB32sII") + + key: Any + compressor: CompressorBase + assert_id_places: frozenset[str] + assert_id_place: str @classmethod def parse_header(cls, buf: bytes | memoryview) -> tuple["RepoObj.ObjHeader | None", str | None]: @@ -97,7 +109,7 @@ def parse_header(cls, buf: bytes | memoryview) -> tuple["RepoObj.ObjHeader | Non return None, f"unsupported object version {hdr.version}" return hdr, None - def __init__(self, key): + def __init__(self, key: Any) -> None: self.key = key # Some commands write new chunks (e.g. rename) but don't take a --compression argument. This duplicates # the default used by those commands who do take a --compression argument. @@ -120,13 +132,13 @@ def id_hash(self, data: bytes) -> bytes: def format( self, id: bytes, - meta: dict, - data: bytes, + meta: dict[str, Any], + data: bytes | memoryview, compress: bool = True, - size: int = None, - ctype: int = None, - clevel: int = None, - ro_type: str = None, + size: int | None = None, + ctype: int | None = None, + clevel: int | None = None, + ro_type: str | None = None, ) -> bytes: assert isinstance(ro_type, str) assert ro_type != ROBJ_DONTCARE @@ -136,6 +148,7 @@ def format( assert isinstance(meta, dict) assert isinstance(data, (bytes, memoryview)) assert compress or size is not None and ctype is not None and clevel is not None + data_compressed: bytes | memoryview if compress: assert size is None or size == len(data) meta, data_compressed = self.compressor.compress(meta, data) @@ -156,7 +169,7 @@ def format( hdr_packed = self.obj_header.pack(*hdr) return hdr_packed + meta_encrypted + data_encrypted - def parse_meta(self, id: bytes, cdata: bytes | memoryview, ro_type: str) -> dict: + def parse_meta(self, id: bytes, cdata: bytes | memoryview, ro_type: str) -> dict[str, Any]: # when calling parse_meta, enough cdata needs to be supplied to contain completely the # meta_len_hdr and the encrypted, packed metadata. it is allowed to provide more cdata. assert isinstance(id, bytes) @@ -187,9 +200,9 @@ def parse( cdata: bytes | memoryview, decompress: bool = True, want_compressed: bool = False, - ro_type: str = None, - assert_id_place: str = None, - ) -> tuple[dict, bytes]: + ro_type: str | None = None, + assert_id_place: str | None = None, + ) -> tuple[dict[str, Any] | None, bytes | None]: """ Parse a repo object into metadata and data (decrypt it, maybe decompress, maybe verify if the chunk plaintext corresponds to the chunk id via assert_id()). @@ -256,7 +269,7 @@ def parse( return meta_compressed if want_compressed else meta, data_compressed if want_compressed else data -def object_validator(repo_objs): +def object_validator(repo_objs: RepoObj) -> Callable[[bytes, bytes | memoryview], bool]: """Return validate(chunk_id, obj): True if obj is the repo object with id chunk_id. obj is an object's header plus its metadata slot. Parsing that slot verifies its tag, which is @@ -272,7 +285,7 @@ def object_validator(repo_objs): hdr_size = RepoObj.obj_header.size overhead = repo_objs.key.PAYLOAD_OVERHEAD # the envelope adds a fixed number of bytes to the payload - def validate(chunk_id, obj): + def validate(chunk_id: bytes, obj: bytes | memoryview) -> bool: try: meta = repo_objs.parse_meta(chunk_id, obj, ro_type=ROBJ_DONTCARE) except (IntegrityErrorBase, msgpack.UnpackException): @@ -296,7 +309,7 @@ class ObjectsNotAuthenticatable(Error): """Objects can not be authenticated with BORG_WORKAROUNDS=authenticated_no_key.""" -def whole_object_authenticator(repo_objs): +def whole_object_authenticator(repo_objs: RepoObj) -> Callable[[bytes, bytes | memoryview], bool]: """Return authenticate(chunk_id, obj): True if obj is the whole repo object with id chunk_id. obj is an object's header, metadata slot and data slot. Parsing it checks that the header's sizes @@ -313,7 +326,7 @@ def whole_object_authenticator(repo_objs): if AUTHENTICATED_NO_KEY and isinstance(repo_objs.key, MACKeyBase): raise ObjectsNotAuthenticatable - def authenticate(chunk_id, obj): + def authenticate(chunk_id: bytes, obj: bytes | memoryview) -> bool: try: repo_objs.parse(chunk_id, obj, decompress=False, want_compressed=True, ro_type=ROBJ_DONTCARE) except (IntegrityErrorBase, msgpack.UnpackException):