refactor!: make rfc folders private

Author: lepture

pyproject.toml

@@ -83,7 +83,7 @@ filterwarnings = ["error"]
 branch = true
 source = ["joserfc"]
 omit = [
-  "src/joserfc/rfc7797/*",
+  "src/joserfc/rfc*/*",
 ]
 
 [tool.coverage.paths]

src/joserfc/_keys.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 import typing as t
 import random
-from .rfc7517.types import AnyKey, KeyParameters, DictKey
-from .rfc7518.oct_key import OctKey
-from .rfc7518.rsa_key import RSAKey
-from .rfc7518.ec_key import ECKey
-from .rfc8037.okp_key import OKPKey
+from ._rfc7517.types import AnyKey, KeyParameters, DictKey
+from ._rfc7518.oct_key import OctKey
+from ._rfc7518.rsa_key import RSAKey
+from ._rfc7518.ec_key import ECKey
+from ._rfc8037.okp_key import OKPKey
 from .errors import (
     MissingKeyError,
     InvalidKeyIdError,

src/joserfc/_rfc7515/__init__.py

@@ -0,0 +1,46 @@
+import typing as t
+from .model import JWSAlgModel, CompactSignature
+from ..errors import (
+    DecodeError,
+    MissingAlgorithmError,
+)
+from ..util import (
+    json_b64encode,
+    json_b64decode,
+    urlsafe_b64encode,
+    urlsafe_b64decode,
+)
+
+
+def sign_compact(obj: CompactSignature, alg: JWSAlgModel, key: t.Any) -> bytes:
+    header_segment = json_b64encode(obj.headers())
+    payload_segment = urlsafe_b64encode(obj.payload)
+    signing_input = header_segment + b"." + payload_segment
+    signature = urlsafe_b64encode(alg.sign(signing_input, key))
+    return signing_input + b"." + signature
+
+
+def verify_compact(obj: CompactSignature, alg: JWSAlgModel, key: t.Any) -> bool:
+    signing_input = obj.segments["header"] + b"." + obj.segments["payload"]
+    try:
+        sig = urlsafe_b64decode(obj.segments["signature"])
+    except (TypeError, ValueError):
+        return False
+    return alg.verify(signing_input, sig, key)
+
+
+def detach_compact_content(value: str) -> str:
+    # https://www.rfc-editor.org/rfc/rfc7515#appendix-F
+    parts = value.split(".")
+    parts[1] = ""
+    return ".".join(parts)
+
+
+def decode_header(header_segment: bytes) -> t.Dict[str, t.Any]:
+    try:
+        protected: t.Dict[str, t.Any] = json_b64decode(header_segment)
+        if "alg" not in protected:
+            raise MissingAlgorithmError()
+    except (TypeError, ValueError):
+        raise DecodeError("Invalid header")
+    return protected

src/joserfc/_rfc7515/compact.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+import typing as t
+import copy
+from .model import (
+    HeaderMember,
+    GeneralJSONSignature,
+    FlattenedJSONSignature,
+)
+from .types import (
+    JSONSignatureDict,
+    GeneralJSONSerialization,
+    FlattenedJSONSerialization,
+)
+from .registry import JWSRegistry
+from ..util import (
+    json_b64encode,
+    json_b64decode,
+    urlsafe_b64encode,
+    urlsafe_b64decode,
+)
+from ..errors import DecodeError
+
+FindKey = t.Callable[[HeaderMember], t.Any]
+
+
+def sign_general_json(
+    members: list[HeaderMember],
+    payload: bytes,
+    registry: JWSRegistry,
+    find_key: FindKey,
+) -> GeneralJSONSerialization:
+    payload_segment = urlsafe_b64encode(payload)
+    signatures: list[JSONSignatureDict] = [
+        sign_json_member(payload_segment, member, registry, find_key) for member in members
+    ]
+    return {
+        "payload": payload_segment.decode("utf-8"),
+        "signatures": signatures,
+    }
+
+
+def sign_flattened_json(
+    member: HeaderMember,
+    payload: bytes,
+    registry: JWSRegistry,
+    find_key: FindKey,
+) -> FlattenedJSONSerialization:
+    payload_segment = urlsafe_b64encode(payload)
+    signature = sign_json_member(payload_segment, member, registry, find_key)
+    data: FlattenedJSONSerialization = {"payload": payload_segment.decode("utf-8"), **signature}
+    return data
+
+
+def sign_json_member(
+    payload_segment: bytes, member: HeaderMember, registry: JWSRegistry, find_key: FindKey
+) -> JSONSignatureDict:
+    headers = member.headers()
+    registry.check_header(headers)
+    alg = registry.get_alg(headers["alg"])
+    key = find_key(member)
+    key.check_use("sig")
+    alg.check_key_type(key)
+    if member.protected:
+        protected_segment = json_b64encode(member.protected)
+    else:
+        protected_segment = b""
+    signing_input = b".".join([protected_segment, payload_segment])
+    signature = urlsafe_b64encode(alg.sign(signing_input, key))
+    rv: JSONSignatureDict = {"signature": signature.decode("utf-8")}
+    if member.protected:
+        rv["protected"] = protected_segment.decode("utf-8")
+    if member.header:
+        rv["header"] = member.header
+    return rv
+
+
+def extract_general_json(value: GeneralJSONSerialization) -> GeneralJSONSignature:
+    payload_segment: bytes = value["payload"].encode("utf-8")
+    try:
+        payload = urlsafe_b64decode(payload_segment)
+    except (TypeError, ValueError):
+        raise DecodeError("Invalid payload")
+
+    signatures: list[JSONSignatureDict] = value["signatures"]
+    members = [__signature_to_member(sig) for sig in signatures]
+    obj = GeneralJSONSignature(members, payload)
+    obj.signatures = signatures
+    obj.segments = {"payload": payload_segment}
+    return obj
+
+
+def __signature_to_member(sig: JSONSignatureDict) -> HeaderMember:
+    member = HeaderMember()
+    if "protected" in sig:
+        protected_segment = sig["protected"]
+        member.protected = json_b64decode(protected_segment)
+    if "header" in sig:
+        member.header = sig["header"]
+    return member
+
+
+def verify_general_json(obj: GeneralJSONSignature, registry: JWSRegistry, find_key: FindKey) -> bool:
+    payload_segment = obj.segments["payload"]
+    for index, signature in enumerate(obj.signatures):
+        member = obj.members[index]
+        if not verify_signature(member, signature, payload_segment, registry, find_key):
+            return False
+    return True
+
+
+def verify_flattened_json(obj: FlattenedJSONSignature, registry: JWSRegistry, find_key: FindKey) -> bool:
+    payload_segment = obj.segments["payload"]
+    assert obj.signature is not None
+    return verify_signature(obj.member, obj.signature, payload_segment, registry, find_key)
+
+
+def verify_signature(
+    member: HeaderMember,
+    signature: JSONSignatureDict,
+    payload_segment: bytes,
+    registry: JWSRegistry,
+    find_key: FindKey,
+) -> bool:
+    headers = member.headers()
+    registry.check_header(headers)
+    alg = registry.get_alg(headers["alg"])
+    key = find_key(member)
+    key.check_use("sig")
+    alg.check_key_type(key)
+    if "protected" in signature:
+        protected_segment = signature["protected"].encode("utf-8")
+    else:
+        protected_segment = b""
+    sig = urlsafe_b64decode(signature["signature"].encode("utf-8"))
+    signing_input = b".".join([protected_segment, payload_segment])
+    return alg.verify(signing_input, sig, key)
+
+
+def detach_json_content(value: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
+    # https://www.rfc-editor.org/rfc/rfc7515#appendix-F
+    rv = copy.deepcopy(value)  # don't alter original value
+    if "payload" in rv:
+        del rv["payload"]
+    return rv

src/joserfc/_rfc7515/json.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+from typing import Any, ClassVar
+from abc import ABCMeta, abstractmethod
+from .types import SegmentsDict, JSONSignatureDict
+from ..errors import InvalidKeyTypeError
+from ..registry import Header
+
+
+class HeaderMember:
+    """A header member of the JSON signature. It is combined with protected header,
+    and unprotected header.
+    """
+
+    def __init__(self, protected: Header | None = None, header: Header | None = None):
+        #: protected header
+        self.protected = protected
+        #: unprotected header
+        self.header = header
+
+    def headers(self) -> Header:
+        rv: Header = {}
+        if self.protected:
+            rv.update(self.protected)
+        if self.header:
+            rv.update(self.header)
+        return rv
+
+    def set_kid(self, kid: str) -> None:
+        if self.header is None:
+            self.header = {}
+        self.header["kid"] = kid
+
+
+class CompactSignature:
+    """JSON Web Signature object for compact mode. This object is used to
+    represent the JWS instance.
+    """
+
+    def __init__(self, protected: Header, payload: bytes):
+        #: protected header
+        self.protected = protected
+        #: payload content in bytes
+        self.payload = payload
+        self.segments: SegmentsDict = {}
+
+    def headers(self) -> Header:
+        """Returns protected header values in dict."""
+        return self.protected
+
+    def set_kid(self, kid: str) -> None:
+        self.protected["kid"] = kid
+
+
+class FlattenedJSONSignature:
+    """JSON Signature object that represents a flattened JSON serialization."""
+
+    #: mark it as flattened
+    flattened: ClassVar[bool] = True
+
+    def __init__(self, member: HeaderMember, payload: bytes):
+        #: the only header member
+        self.member: HeaderMember = member
+        #: payload content in bytes
+        self.payload: bytes = payload
+        self.signature: JSONSignatureDict | None = None
+        self.segments: SegmentsDict = {}
+
+    @property
+    def members(self) -> list[HeaderMember]:
+        """A list of header members. For flattened JSON serialization, there will
+        be only one header member."""
+        return [self.member]
+
+    def headers(self) -> Header:
+        """Header values in dict."""
+        return self.member.headers()
+
+
+class GeneralJSONSignature:
+    """JSON Signature object that represents a general JSON serialization."""
+
+    #: mark it as not flattened (general)
+    flattened: ClassVar[bool] = False
+
+    def __init__(self, members: list[HeaderMember], payload: bytes):
+        #: a list of header members
+        self.members: list[HeaderMember] = members
+        #: payload content in bytes
+        self.payload: bytes = payload
+        self.signatures: list[JSONSignatureDict] = []
+        self.segments: SegmentsDict = {}
+
+
+class JWSAlgModel(object, metaclass=ABCMeta):
+    """Interface for JWS algorithm. JWA specification (RFC7518) SHOULD
+    implement the algorithms for JWS with this base implementation.
+    """
+
+    name: str
+    description: str
+    recommended: bool = False
+    key_type = "oct"
+    algorithm_type = "JWS"
+    algorithm_location = "sig"
+
+    def check_key_type(self, key: Any) -> None:
+        if key.key_type != self.key_type:
+            raise InvalidKeyTypeError(f"Algorithm '{self.name}' requires '{self.key_type}' key")
+
+    @abstractmethod
+    def sign(self, msg: bytes, key: Any) -> bytes:
+        """Sign the text msg with a private/sign key.
+
+        :param msg: message bytes to be signed
+        :param key: private key to sign the message
+        :return: bytes
+        """
+
+    @abstractmethod
+    def verify(self, msg: bytes, sig: bytes, key: Any) -> bool:
+        """Verify the signature of text msg with a public/verify key.
+
+        :param msg: message bytes to be signed
+        :param sig: result signature to be compared
+        :param key: public key to verify the signature
+        :return: boolean
+        """