diff --git a/README.md b/README.md index 2ab8679..fe5a0bc 100644 --- a/README.md +++ b/README.md @@ -36,10 +36,12 @@ format defines no smaller payload limit. The null-terminated domain carries no length of its own, so the protocol alone is not an application input limit for the complete envelope. -This package validates that the declared payload length agrees with the bytes -present before it sizes or copies the payload. A large declaration without -the corresponding bytes is malformed and is rejected without allocating the -declared size. A matching large payload is not malformed merely because it is +This package compares the declared payload length with the bytes actually +present before it sizes or copies the payload, so a large declaration without +the corresponding bytes is refused without ever allocating the declared size. +That refusal is `ParseStatus.BYTE_COUNT_MISMATCH` on a whole buffer read and +`ParseStatus.UNEXPECTED_END` on a framed one, for the reason given under +reading below. A matching large payload is not malformed merely because it is large, and parsing work and memory use scale with the bytes actually present. The domain ends at a zero terminator rather than at a declared length, so a @@ -55,19 +57,24 @@ cannot read moves the fault to the consumer. A `Creator` refuses a domain longer than `MAXIMUM_DOMAIN_LENGTH` when the caller supplies it, before any signing work is done, and the writer refuses one that reached an OWID by any other route when the OWID is serialised. Both raise `OwidError` naming the -maximum, as the parse does. +maximum, because a domain that long is a fault in the calling code rather than +data arriving from outside. A read reports the same finding as +`ParseStatus.INVALID_DOMAIN_ENCODING` instead of raising, as there the domain +came from whoever sent the bytes. The in-memory APIs remain subject to Python object, address-space and available-memory limits. Applications accepting untrusted OWIDs must choose limits suitable for their use case and enforce them before buffering the binary form or decoding Base64. An implementation capacity failure or an -application policy rejection is distinct from an invalid OWID. +application policy rejection is distinct from an invalid OWID, which is why +`ParseStatus.IMPLEMENTATION_CAPACITY_EXCEEDED` is a status of its own and the +same bytes may be readable somewhere with more room. -For transport input, limit the complete HTTP body or encoded envelope; allow -for the domain and other OWID fields as well as the payload. After parsing, -`len(owid.payload)` reports the actual payload size without another copy and -can be used for downstream policy. The parser cannot choose either limit on -behalf of the application. +For transport input, limit the complete HTTP body or encoded envelope, and +allow for the domain and other OWID fields as well as the payload. After a +successful read, `len(result.owid.payload)` reports the actual payload size +without another copy and can be used for downstream policy. The reader cannot +choose either limit on behalf of the application. ## Installation @@ -88,6 +95,9 @@ python -m pip install -e . ## Usage +Create a creator that holds the signing keys, create a signed OWID, serialize +it, then read it back later and verify it with the public key. + ```python from owid import Creator, Crypto, Owid @@ -95,45 +105,210 @@ from owid import Creator, Crypto, Owid crypto = Crypto.new() creator = Creator("example.com", crypto) -# Create and sign an OWID with a payload. -owid = creator.sign_string("Hello World") +# Creating and signing are one step, so an OWID never exists unsigned. +owid = creator.create_string("Hello World") # Serialize to base 64 for storage or transmission. encoded = owid.as_base64() -# Later, or elsewhere, decode and verify with the creator public key. -copy = Owid.from_base64(encoded) -public_pem = crypto.public_key_pem() -assert copy.verify_with_public_key(public_pem, []) +# Later, or elsewhere, read it back. Input from outside may be anything at +# all, so reading answers with a result rather than raising. +result = Owid.parse(encoded) +if result: + public_pem = crypto.public_key_pem() + assert result.owid.verify_with_public_key(public_pem, []) +else: + # result.status names which of the expected problems it was, for example + # ParseStatus.INVALID_BASE64 or ParseStatus.BYTE_COUNT_MISMATCH. + reason = result.status.value ``` -OWIDs chain together. To sign an OWID with another OWID covered by the same -signature, pass the others when signing and the same others, in the same -order, when verifying. +OWIDs chain together. Create one that covers others, and pass the same others, +in the same order, when verifying. ```python -root = creator.sign_string("root") -party = Owid(payload=b"party") -creator.sign_with_others(party, [root]) +root = creator.create_string("root") +party = creator.create(b"party", [root]) assert party.verify_with_crypto(crypto, [root]) assert not party.verify_with_crypto(crypto, []) ``` +Where the difference between a signature that does not match and a check that +could not be made changes what your code should do, ask for the status rather +than a true or false answer. A key that cannot be read is reported as a fault +in the key and never as a forgery. + +```python +from owid import SignatureStatus + +status = owid.signature_status(crypto.public_key_pem()) +if status is SignatureStatus.SIGNATURE_VALID: + pass # Genuine. +elif status is SignatureStatus.SIGNATURE_INVALID: + pass # The only status meaning the identifier should be distrusted. +else: + # INVALID_KEY, VERIFICATION_ERROR and the rest mean the question could + # not be answered, which is an operational fault rather than an attack. + pass +``` + +## How an OWID comes into being + +An OWID is only worth anything because it is signed, so a caller cannot build +one. An instance arrives by exactly two routes. + +1. Reading bytes that were already a complete OWID, with `Owid.parse`, + `Owid.parse_bytes` or `Owid.parse_prefix`. +2. `Creator.create` and `Creator.create_string`, which own the version, the + domain, the date and the signature, and hand back a finished OWID. + +Python cannot make a constructor private, so calling `Owid()` raises +`OwidError` naming the two routes instead. The payload and the signature are +handed out as copies and the fields are read only properties, because a parsed +OWID's signature covers its fields as they arrived, so code that could change +them afterwards would hold something whose signature no longer describes it. +There is no way to sign an OWID that already exists, as an unsigned OWID is +indistinguishable from a signed one to the code downstream of it and the +difference only surfaces later when a verification fails somewhere nobody is +watching. + +## Reading data that may not be an OWID + +An OWID is read from whatever a caller was handed, which on a public end point +means anything at all, so being malformed is an ordinary outcome rather than an +exceptional one. The parse methods report it instead of raising, because +raising costs the construction and unwinding of an exception for every bad +input and whoever sends the data chooses how often that happens. + +Every read hands back a `ParseResult`, which is truthy on success and reports +the same three facts. + +1. `result.ok`, whether it worked. +2. `result.owid`, the OWID on success and `None` on failure. +3. `result.status`, a `ParseStatus` naming the reason, which is + `ParseStatus.PARSED` on success. + +A result also carries `result.consumed`, the number of bytes the envelope +occupied, which a caller reading several OWIDs from one buffer uses to reach +the next one. Nothing is consumed when an envelope is refused, because a half +read one leaves a caller somewhere it cannot reason about. + +The reasons a read can give are named by `ParseStatus`. + +| Status | Meaning | +| ------ | ------- | +| `PARSED` | The bytes form a structurally valid OWID. This says nothing about the signature. | +| `MISSING_INPUT` | Nothing was supplied to parse. | +| `INVALID_INPUT_TYPE` | The input arrived in a form the surface cannot read, such as bytes where a base 64 string was wanted. | +| `INVALID_BASE64` | The string is not valid base 64, so there are no bytes to read. | +| `UNSUPPORTED_VERSION` | The first byte names a version this implementation does not know. | +| `UNEXPECTED_END` | The data stopped in the middle of a field. | +| `INVALID_DOMAIN_ENCODING` | The creator domain is not terminated, or is longer than the published maximum. | +| `BYTE_COUNT_MISMATCH` | The declared payload byte count disagrees with the bytes actually present. | +| `IMPLEMENTATION_CAPACITY_EXCEEDED` | The envelope is consistent but larger than this runtime can hold, so the same bytes may be readable elsewhere. | +| `ABSENT_NODE` | The one byte marker standing for a node that is not there. | +| `MALFORMED_ENVELOPE` | Malformed in a way none of the above describes. | + +These names are the cross language vocabulary, so a failure means the same +thing whichever implementation read the bytes, and code matching on a status +never has to match on message text. + +### The whole buffer contract and the framed contract + +`Owid.parse` and `Owid.parse_bytes` require the value to be one whole OWID and +nothing else, because on those surfaces there is nothing else the bytes after +the envelope could belong to. The declared payload must leave exactly the +signature, so a byte after it is `ParseStatus.BYTE_COUNT_MISMATCH`. + +`Owid.parse_prefix` reads one OWID from the front of a buffer that may carry +more after it and leaves the rest alone, because what follows may be the next +envelope rather than rubbish. It needs only the declared payload and the +signature to be present, and says nothing about the bytes beyond them. A frame +whose declaration runs past the bytes supplied is therefore +`ParseStatus.UNEXPECTED_END`, being data that stopped early rather than a +declaration disagreeing with data that is all present, which is the answer a +caller reading a source still arriving needs so that it can wait for more bytes +instead of giving up. + +### The marker for a node that is absent + +A single zero byte, written by `Owid.empty_to_buffer`, stands for an optional +OWID that is not there. Both reads report it as `ParseStatus.ABSENT_NODE` and +neither hands back an OWID, because the marker carries no domain, date, payload +or signature and so can never verify, and reading one as an identifier would be +the one way an instance with no signature could reach calling code. It is not +an unknown version, as version 0 is supported and meaningful, and it is not a +malformed frame either. + +A framed read counts the marker's one byte as consumed, so a caller walking a +run of frames steps over an absent node and reaches the next envelope. A whole +buffer read consumes nothing, as there the marker on its own is the whole of +what was supplied. + +```python +from owid import Owid, ParseStatus + +# A buffer holding one OWID, a node that is absent, then another OWID. +buffer = bytearray() +creator.create_string("first").to_buffer(buffer) +Owid.empty_to_buffer(buffer) +creator.create_string("second").to_buffer(buffer) + +data = bytes(buffer) +payloads = [] +while data: + frame = Owid.parse_prefix(data) + if frame: + payloads.append(frame.owid.payload_as_string()) + elif frame.status is not ParseStatus.ABSENT_NODE: + break + data = data[frame.consumed:] + +assert payloads == ["first", "second"] +``` + +Reading is not verification. A successfully read OWID is structurally valid and +nothing more, and whether its signature is genuine is a separate question with +a separate answer. + ## Interface `Owid` -- `from_base64(value)` and `from_byte_array(buffer)` parse a signed OWID. -- `as_base64()` and `as_byte_array()` serialize a signed OWID. +- `parse(value)` reads a complete OWID from its base 64 form. +- `parse_bytes(buffer)` reads a complete OWID from a buffer holding exactly + one. +- `parse_prefix(buffer)` reads one OWID from the front of a buffer that may + carry more after it. +- `version`, `domain`, `date`, `payload` and `signature` are read only + properties. +- `as_base64()` and `as_byte_array()` serialize a signed OWID, and + `to_buffer(buffer)` appends it to a `bytearray`. +- `empty_to_buffer(buffer)` writes the one byte marker for a node that is not + there. - `payload_as_string()` decodes the payload as UTF-8, replacing invalid bytes. - `payload_as_printable()` returns the payload as lower case hexadecimal. - `payload_as_base64()` returns the payload as a base 64 string. - `verify_with_crypto(crypto, others)` and `verify_with_public_key(public_pem, others)` return True if the signature is valid. Pass an empty list for `others` when the OWID was signed on its own. +- `signature_status(public_pem, others)` answers the same question with a + `SignatureStatus`, which keeps a signature that does not match apart from a + check that could not be made at all. - `age_minutes()` returns the whole minutes elapsed since creation. +`ParseResult` + +- `ok`, `owid`, `status` and `consumed`, described under reading above. The + result is truthy when `ok` is True, so `if result:` reads the way Python + reads. + +`ParseStatus` and `SignatureStatus` + +- The named reasons a read or a signature check reports. Both are enumerations + whose `value` is the cross language name, for example `"ByteCountMismatch"`. + `Crypto` - `new()` generates a P-256 key pair. @@ -141,7 +316,9 @@ assert not party.verify_with_crypto(crypto, []) - `new_verify_only(public_pem)` imports an SPKI public key PEM. - `sign_byte_array(data)` returns the 64 byte signature. - `verify_byte_array(data, signature)` returns True if the signature is valid. -- `subject_public_key_info()` and `private_key_pem()` export the keys as PEM. +- `subject_public_key_info()` and `private_key_pem()` export the keys as PEM, + and `public_key_pem()` is an alias of the first of those. +- `can_sign()` and `can_verify()` report which keys the instance holds. An empty or whitespace PEM is rejected with a clear message rather than an opaque crypto error. @@ -151,9 +328,9 @@ opaque crypto error. - `Creator(domain, crypto)` binds a domain to a signing crypto instance. - `from_configuration(configuration)` builds a creator from a domain and a private key PEM. -- `sign(owid)` and `sign_with_others(owid, others)` set the domain, date, and - version, then sign. -- `sign_string(value)` and `sign_bytes(value)` create and sign a new OWID. +- `create(value, others)` creates and signs a new OWID carrying the bytes, + covering any others with the same signature. +- `create_string(value)` does the same with the UTF-8 bytes of a string. `endpoints` @@ -189,8 +366,19 @@ this OWID without its signature, followed by the complete bytes, including the signature, of each other OWID in the order given. The same others in the same order must be supplied to verify as were supplied to sign. -An empty OWID is written as a single byte with value 0 and acts as a marker -for an absent optional OWID inside a larger byte array. +A single byte with value 0 is the marker for an absent optional OWID inside a +larger byte array, written by `Owid.empty_to_buffer` and reported by both reads +as `ParseStatus.ABSENT_NODE`, covered under reading above. + +The two reads differ in three answers and agree everywhere else. A whole buffer +read requires the declared payload to leave exactly the signature, so a byte +after it is `ParseStatus.BYTE_COUNT_MISMATCH`, while a framed read requires +only that the payload and the signature are present and says nothing about what +follows. A frame whose declared payload runs past the bytes supplied is +`ParseStatus.UNEXPECTED_END`, so `ParseStatus.BYTE_COUNT_MISMATCH` is reachable +only on the whole buffer read where every byte is present by definition. A +framed read counts the marker's one byte as consumed and a whole buffer read +counts nothing. Base 64 decoding accepts input with or without trailing padding. Encoding always emits padding. @@ -199,7 +387,11 @@ always emits padding. The tests use the standard library `unittest` and exercise the canonical wire vectors, the cross language signed fixtures, the signing path, and the unit -behaviour of each module. Run them from the repository root. +behaviour of each module. `tests/test_parse_contract.py` holds the cross +language status matrix, being the reasons a read reports and the proof that an +OWID cannot be held unsigned. `tests/test_readme.py` runs the Python examples +in this file in the order they appear, so documentation naming a method that +does not exist fails the build. Run them from the repository root. ``` python -m unittest discover diff --git a/owid/__init__.py b/owid/__init__.py index 5e3ddf7..3892e7b 100644 --- a/owid/__init__.py +++ b/owid/__init__.py @@ -33,6 +33,8 @@ from .error import OwidError from .io import SIGNATURE_LENGTH from .owid import Owid +from .parse import ParseResult +from .status import ParseStatus, SignatureStatus from .version import DEFAULT_VERSION, Version __all__ = [ @@ -41,6 +43,12 @@ "Crypto", "OwidError", "Owid", + # A caller cannot act on a read without naming the reason it carries, so + # the result and both status vocabularies sit beside the type they + # describe rather than in a module a reader has to go looking for. + "ParseResult", + "ParseStatus", + "SignatureStatus", "Version", "DEFAULT_VERSION", "SIGNATURE_LENGTH", diff --git a/owid/creator.py b/owid/creator.py index bc0346a..f6f4d5d 100644 --- a/owid/creator.py +++ b/owid/creator.py @@ -105,40 +105,53 @@ def crypto(self) -> Crypto: """Used to sign OWIDs from this creator.""" return self._crypto - def sign(self, owid: Owid) -> None: + def _sign(self, owid: Owid) -> None: """Signs the OWID provided, setting the domain to the creator domain, the date to the current time, and the version to the current version, then signing it.""" - self.sign_with_others(owid, []) + self._sign_with_others(owid, []) - def sign_with_others(self, owid: Owid, others: Sequence[Owid]) -> None: + def _sign_with_others(self, owid: Owid, others: Sequence[Owid]) -> None: """Signs the OWID provided together with the other OWIDs provided. The same others, in the same order, must be passed when verifying. """ - owid.version = DEFAULT_VERSION - owid.domain = self._domain + owid._version = DEFAULT_VERSION + owid._domain = self._domain # Truncate to whole minutes so the in-memory date matches the value # written to the wire format. This keeps a signed OWID equal to a copy # decoded from its bytes, and ensures signing and verification operate # on the same minute precise value. - owid.date = datetime.now(timezone.utc).replace(second=0, microsecond=0) + owid._date = datetime.now(timezone.utc).replace(second=0, microsecond=0) data = owid.data_for_crypto(others) - owid.signature = self._crypto.sign_byte_array(data) - if len(owid.signature) != SIGNATURE_LENGTH: + owid._signature = self._crypto.sign_byte_array(data) + if len(owid._signature) != SIGNATURE_LENGTH: raise OwidError( "signature length '{0}' not compatible with '{1}' OWID " - "signature length".format(len(owid.signature), SIGNATURE_LENGTH) + "signature length".format(len(owid._signature), SIGNATURE_LENGTH) ) - def sign_string(self, value: str) -> Owid: - """Creates a new signed OWID for the creator containing the string as - the payload.""" - return self.sign_bytes(value.encode("utf-8")) + def create_string(self, value: str) -> Owid: + """Creates and signs a new OWID carrying the string as its payload.""" + if value is None: + raise OwidError("a payload is required") + return self.create(value.encode("utf-8")) - def sign_bytes(self, value: bytes) -> Owid: - """Creates a new signed OWID for the creator containing the bytes as - the payload.""" - owid = Owid(payload=value) - self.sign(owid) + def create(self, value: bytes, others: Sequence[Owid] = ()) -> Owid: + """Creates and signs a new OWID carrying the bytes as its payload. + + This is one of only two ways an OWID reaches calling code, the other + being a successful parse. The creator owns the version, the domain, + the date and the signature; a caller supplies the payload and nothing + else, so there is no moment at which a partly built OWID exists for + anyone to hold or pass on. + + Any others are covered by the signature so that a tree can be verified + as a whole, and the same others in the same order must be passed when + verifying. + """ + if value is None: + raise OwidError("a payload is required") + owid = Owid._create(payload=bytes(value)) + self._sign_with_others(owid, list(others)) return owid diff --git a/owid/error.py b/owid/error.py index df553e0..9db5667 100644 --- a/owid/error.py +++ b/owid/error.py @@ -15,15 +15,23 @@ # **************************************************************************** """The error type raised across the package. -A single exception type carries a human readable message. It is raised for -unsupported versions, malformed buffers, key problems, and the other failures -that can occur when creating, reading, signing, or verifying OWIDs. +A single exception type carries a human readable message. It is raised where +the fault lies in the calling code or in the local key material, being a +domain or payload that cannot be written, a key that cannot be imported or +exported, an attempt to construct an OWID directly, and a version the writer +does not know. + +Reading data that came from outside does not raise. Bytes that are not an OWID +are an ordinary outcome, so the parse surfaces answer with a ParseResult +carrying a ParseStatus, and a signature that cannot be judged is reported as a +SignatureStatus rather than as an exception. """ from __future__ import annotations class OwidError(Exception): - """Raised when an OWID can not be created, read, signed, or verified.""" + """Raised when an OWID can not be created, written, signed, or verified, + and never for external data that turns out not to be an OWID.""" pass diff --git a/owid/io.py b/owid/io.py index da35a26..0e152cb 100644 --- a/owid/io.py +++ b/owid/io.py @@ -18,6 +18,10 @@ The format uses little endian unsigned 32 bit integers, null terminated strings, and a fixed 64 byte signature. The date is stored as the count of hours (version 1) or minutes (versions 2 and 3) since the base date. + +The write helpers here are the ones an OWID is serialized with. The Reader is +not the public read, because external data is read by owid.parse, which walks +the bytes by index and reports a ParseStatus instead of raising. """ from __future__ import annotations @@ -47,7 +51,12 @@ class Reader: - """Sequential reader over a byte buffer.""" + """Sequential reader over a byte buffer, raising on anything malformed. + + Kept for the tests that assert those messages, and reached only through + the private raising route on Owid. Callers read external data with + Owid.parse, Owid.parse_bytes or Owid.parse_prefix instead. + """ def __init__(self, buffer: bytes) -> None: self._buffer = buffer diff --git a/owid/owid.py b/owid/owid.py index 486e839..1126986 100644 --- a/owid/owid.py +++ b/owid/owid.py @@ -24,34 +24,20 @@ from __future__ import annotations import base64 -import binascii from datetime import datetime, timezone -from typing import List, Optional, Sequence +from typing import TYPE_CHECKING, Optional, Sequence from . import io from .crypto import Crypto from .error import OwidError from .version import DEFAULT_VERSION, Version - -def _decode_base64(value: str) -> bytes: - """Decodes a standard alphabet base 64 string with or without padding. - - The encoded OWIDs occur both with and without padding, so reading must - accept both. Missing padding is added before decoding. - """ - cleaned = value.strip() - remainder = len(cleaned) % 4 - if remainder == 2: - cleaned += "==" - elif remainder == 3: - cleaned += "=" - elif remainder == 1: - raise OwidError("base 64 decoding failed because the length is invalid") - try: - return base64.b64decode(cleaned, validate=True) - except (binascii.Error, ValueError) as exc: - raise OwidError("base 64 decoding failed because {0}".format(exc)) +if TYPE_CHECKING: + # Only for the annotations below. Importing parse here at run time would + # be a cycle, because parse imports this module to build the OWID it hands + # back, so both names are resolved by the type checker alone. + from .parse import ParseResult + from .status import SignatureStatus def _encode_base64(value: bytes) -> str: @@ -59,8 +45,29 @@ def _encode_base64(value: bytes) -> str: return base64.b64encode(value).decode("ascii") +#: Passed to Owid.__init__ by the two paths allowed to build one. Python +#: cannot make a constructor package private, so the boundary is kept by +#: asking for something a caller outside this package has no reason to have. +_INTERNAL = object() + + class Owid: - """OWID structure which can be used as a node in a tree.""" + """OWID structure which can be used as a node in a tree. + + An OWID is a claim about who created some data and when, and it is only + worth anything because it is signed. A caller therefore cannot build one. + An instance arrives either from parsing bytes that were already a complete + OWID, or from a Creator that signs one into existence. There is + deliberately no way to assemble a half made one, because an unsigned OWID + is indistinguishable from a signed one to the code downstream of it, and + the difference only surfaces later when a verification fails somewhere + nobody is watching. + + The payload and signature are handed out as copies for the same reason: a + parsed OWID's signature covers its fields as they arrived, so code that + could change them afterwards would hold something whose signature no + longer describes it. + """ def __init__( self, @@ -69,30 +76,110 @@ def __init__( date: Optional[datetime] = None, payload: bytes = b"", signature: bytes = b"", + _token: object = None, ) -> None: - #: The byte version of the OWID. - self.version = version - #: Domain associated with the creator. - self.domain = domain - #: The date and time to the nearest minute in UTC of the creation. - self.date = date if date is not None else datetime.now(timezone.utc) - #: Bytes that form the payload. - self.payload = payload - #: Signature for this OWID and any others provided when signing. - self.signature = signature + if _token is not _INTERNAL: + raise OwidError( + "an OWID cannot be constructed directly. Use " + "Owid.parse or Owid.parse_bytes to read " + "one, or Creator.create to sign a new one" + ) + self._version = version + self._domain = domain + self._date = date if date is not None else datetime.now(timezone.utc) + self._payload = bytes(payload) + self._signature = bytes(signature) + + @property + def version(self) -> Version: + """The byte version of the OWID.""" + return self._version + + @property + def domain(self) -> str: + """Domain associated with the creator.""" + return self._domain + + @property + def date(self) -> datetime: + """The date and time to the nearest minute in UTC of the creation.""" + return self._date + + @property + def payload(self) -> bytes: + """Bytes that form the payload. + + Read only, and bytes rather than a mutable buffer, because the + signature covers the payload as it arrived. Code able to change it + would hold something whose signature no longer describes it. + """ + return self._payload + + @property + def signature(self) -> bytes: + """Signature for this OWID and any others provided when signing.""" + return self._signature + + @classmethod + def _create( + cls, + version: Version = DEFAULT_VERSION, + domain: str = "", + date: Optional[datetime] = None, + payload: bytes = b"", + signature: bytes = b"", + ) -> "Owid": + """Builds an instance from fields a parser or creator has validated.""" + return cls( + version=version, + domain=domain, + date=date, + payload=payload, + signature=signature, + _token=_INTERNAL, + ) + + @classmethod + def parse(cls, value) -> "ParseResult": + """Reads a complete OWID from its base 64 form. + + Returns a result that is truthy on success and carries the OWID only + then, with a named reason either way, so ``if result:`` reads the way + Python reads. Malformed input is an ordinary outcome rather than an + exception, because the data comes from outside and whoever sends it + chooses how often it is wrong. + """ + from .parse import parse_base64 + + return parse_base64(value) @classmethod - def from_base64(cls, value: str) -> "Owid": - """Creates an OWID from a base 64 encoded string. + def parse_prefix(cls, buffer) -> "ParseResult": + """Reads one OWID from the start of a buffer that may hold more. - Raises OwidError if the string is not valid base 64 or the bytes do - not form a valid OWID. + The framed read. What follows the envelope is left alone, because it + may be the next one rather than rubbish, and the result says how many + bytes this envelope occupied so a caller can walk a run of them. """ - return cls.from_byte_array(_decode_base64(value)) + from .parse import parse_prefix + + return parse_prefix(buffer) @classmethod - def from_byte_array(cls, buffer: bytes) -> "Owid": - """Creates an OWID from its binary form. + def parse_bytes(cls, buffer) -> "ParseResult": + """Reads a complete OWID from a buffer holding exactly one.""" + from .parse import parse_bytes + + return parse_bytes(buffer) + + @classmethod + def _from_byte_array_or_raise(cls, buffer: bytes) -> "Owid": + """The raising counterpart of parse_bytes, over the low level reader. + + Private, and not the way to read external data, because bytes that + are not an OWID are an ordinary outcome and parse_bytes reports the + reason rather than raising. This route survives so the tests can + assert the messages the low level reader gives. The buffer must hold exactly one OWID, ending with the 64 byte signature. Raises OwidError if the first byte is not a known @@ -108,12 +195,12 @@ def _from_reader(cls, reader: "io.Reader") -> "Owid": """Creates an OWID by reading the next fields from the reader.""" version = Version.from_byte(reader.read_byte()) if version == Version.EMPTY: - return cls(version=version) + return cls._create(version=version) domain = reader.read_string() date = reader.read_date(version) payload = reader.read_byte_array() signature = reader.read_signature() - return cls( + return cls._create( version=version, domain=domain, date=date, @@ -207,6 +294,43 @@ def verify_with_public_key( crypto = Crypto.new_verify_only(public_pem) return self.verify_with_crypto(crypto, others) + def signature_status( + self, public_pem: str, others: Optional[Sequence["Owid"]] = None + ) -> "SignatureStatus": + """Says whether the signature is genuine, or why that could not be + decided. + + Only two of the answers are about the signature. The rest say the + question could not be answered, which is a different thing and must + never be reported as a forgery. A key that cannot be decoded leaves the + signature unjudged, and a caller acting on "invalid" would reject good + identifiers during an outage. On 30 August 2026 the key endpoints + served PEM a strict parser rejects and every offline verification + failed, with the keys and the identifiers both fine. + """ + from .status import SignatureStatus + + if not public_pem: + return SignatureStatus.KEY_UNAVAILABLE + if len(self._signature) != io.SIGNATURE_LENGTH: + return SignatureStatus.INVALID_SIGNATURE_LENGTH + try: + crypto = Crypto.new_verify_only(public_pem) + except Exception: + # The key is the thing at fault, not the identifier. + return SignatureStatus.INVALID_KEY + try: + data = self.data_for_crypto(others if others is not None else []) + matched = crypto.verify_byte_array(data, self._signature) + except Exception: + # The provider failed on inputs that were themselves fine. + return SignatureStatus.VERIFICATION_ERROR + return ( + SignatureStatus.SIGNATURE_VALID + if matched + else SignatureStatus.SIGNATURE_INVALID + ) + def __str__(self) -> str: """Formats the OWID as a base 64 string.""" return self.as_base64() diff --git a/owid/parse.py b/owid/parse.py new file mode 100644 index 0000000..6a61963 --- /dev/null +++ b/owid/parse.py @@ -0,0 +1,249 @@ +# **************************************************************************** +# Copyright 2026 51 Degrees Mobile Experts Limited (51degrees.com) +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# **************************************************************************** + +"""Reading an OWID from bytes without raising for malformed input.""" + +import base64 +import binascii +from datetime import timedelta +from typing import TYPE_CHECKING, NamedTuple, Optional + +from .io import BASE_DATE, MAXIMUM_DOMAIN_LENGTH, SIGNATURE_LENGTH +from .status import ParseStatus +from .version import Version + +if TYPE_CHECKING: + # Only for the annotation below. Importing owid here at run time would be + # a cycle, because this module builds the OWID it hands back. + from .owid import Owid + + +class ParseResult(NamedTuple): + """What a parse produced, and why. + + Truthy on success, so ``if result:`` reads naturally, while the value and + the reason stay available for code that needs to say which of the expected + problems it was. + """ + + #: True when the bytes were a complete, structurally valid OWID. + ok: bool + + #: The OWID on success, otherwise None. + owid: Optional["Owid"] + + #: PARSED on success, otherwise the specific reason. + status: ParseStatus + + #: How many bytes the envelope occupied. Only meaningful for a framed + #: read, where a caller advances by this much to reach the next one. + consumed: int = 0 + + def __bool__(self) -> bool: + return self.ok + + +def _failed(status: ParseStatus) -> ParseResult: + return ParseResult(False, None, status, 0) + + +def parse_base64(value) -> ParseResult: + """Reads a complete OWID from its base 64 form. + + The value may be anything at all: this is external data, and failing to be + an OWID is an ordinary outcome rather than an error. + """ + if value is None or value == "": + return _failed(ParseStatus.MISSING_INPUT) + if not isinstance(value, str): + return _failed(ParseStatus.INVALID_INPUT_TYPE) + # Encoded OWIDs occur both with and without padding, so reading accepts + # both. Missing padding is added rather than the value being refused, + # because an unpadded encoding is a normal way to carry one, not a fault. + cleaned = value.strip() + remainder = len(cleaned) % 4 + if remainder == 1: + return _failed(ParseStatus.INVALID_BASE64) + if remainder == 2: + cleaned += "==" + elif remainder == 3: + cleaned += "=" + try: + buffer = base64.b64decode(cleaned, validate=True) + except (binascii.Error, ValueError): + return _failed(ParseStatus.INVALID_BASE64) + return parse_bytes(buffer) + + +def parse_prefix(buffer) -> ParseResult: + """Reads one OWID from the start of a buffer that may hold more after it. + + This is the framed contract. It differs from parse_bytes in exactly one + place: a whole buffer knows where the envelope ends, so the declared + payload must leave exactly the signature, while here what follows may be + the next envelope rather than rubbish, so the declaration and the + signature need only be present. + + The result carries how many bytes the envelope occupied, so a caller walks + a run of them by slicing: + + while data: + result = Owid.parse_prefix(data) + if not result: + break + use(result.owid) + data = data[result.consumed:] + + Nothing is consumed when an envelope is refused, because a half read one + leaves a caller somewhere it cannot reason about. + """ + return _parse(buffer, exact=False) + + +def parse_bytes(buffer) -> ParseResult: + """Reads a complete OWID from a buffer holding exactly one. + + The buffer must be one whole OWID and nothing else. Data after the + envelope is rejected, because on this surface there is nothing else it + could belong to. + + The buffer is walked by index and every read is checked against what is + left, so a malformed envelope is a comparison that fails rather than an + exception that unwinds. That matters because whoever is sending the data + chooses how often this fails and how large each attempt is. + """ + return _parse(buffer, exact=True) + + +def _parse(buffer, exact: bool) -> ParseResult: + """The one walk both reads share.""" + if buffer is None: + return _failed(ParseStatus.MISSING_INPUT) + if not isinstance(buffer, (bytes, bytearray, memoryview)): + return _failed(ParseStatus.INVALID_INPUT_TYPE) + + data = bytes(buffer) + total = len(data) + if total < 1: + # Nothing was supplied, which is not the same as data that stopped + # part way through a field. + return _failed(ParseStatus.MISSING_INPUT) + + # Imported here rather than at module scope: owid.py imports this module + # for its try_ surfaces, so importing it back at the top would be a cycle. + from .owid import Owid + + at = 1 + try: + version = Version.from_byte(data[0]) + except Exception: + return _failed(ParseStatus.UNSUPPORTED_VERSION) + + if version == Version.EMPTY: + # The marker stands for an absent node inside a stream. It is not an + # OWID: it carries no domain, date, payload or signature, so it can + # never verify, and no value is handed back. A framed read still moves + # past its one byte, so a caller walking a run of frames can skip an + # absent node deliberately rather than being unable to tell one from a + # malformed frame. + return ParseResult(False, None, ParseStatus.ABSENT_NODE, + 1 if not exact else 0) + + # The domain, terminated by a zero byte and no longer than the published + # maximum. + start = at + limit = min(total, start + MAXIMUM_DOMAIN_LENGTH + 1) + domain = None + while at < limit: + if data[at] == 0: + domain = data[start:at].decode("ascii", errors="replace") + at += 1 + break + at += 1 + if domain is None: + # Either the buffer ended inside the domain, or the domain ran past + # the maximum without terminating. The second is a domain that cannot + # be valid rather than data that merely stopped. + if at >= total and (at - start) <= MAXIMUM_DOMAIN_LENGTH: + return _failed(ParseStatus.UNEXPECTED_END) + return _failed(ParseStatus.INVALID_DOMAIN_ENCODING) + + # The date, whose width depends on the version. + if version == Version.VERSION1: + if total - at < 2: + return _failed(ParseStatus.UNEXPECTED_END) + hours = (data[at] << 8) | data[at + 1] + at += 2 + date = BASE_DATE + timedelta(hours=hours) + else: + if total - at < 4: + return _failed(ParseStatus.UNEXPECTED_END) + minutes = int.from_bytes(data[at:at + 4], "little") + at += 4 + date = BASE_DATE + timedelta(minutes=minutes) + + if total - at < 4: + return _failed(ParseStatus.UNEXPECTED_END) + declared = int.from_bytes(data[at:at + 4], "little") + at += 4 + + # The declaration is the sender's claim about a payload not yet read, so + # it is compared with what is actually present before anything is sized by + # it. The subtraction is signed, so a buffer with fewer bytes left than a + # signature needs gives a negative count, which can never equal a + # declaration. Reporting that as a truncation would name a different fault + # for the same evidence: what is certain is that the declared payload + # cannot leave exactly the signature the version requires. + # + # The two contracts differ here, and only here. A whole buffer knows the + # envelope boundary, so the declaration must leave exactly the signature + # and no more. A framed read does not: what follows may be the next + # envelope, so it needs the declaration and the signature to be present + # and says nothing about the rest. + present = (total - at) - SIGNATURE_LENGTH + if exact: + if present != declared: + return _failed(ParseStatus.BYTE_COUNT_MISMATCH) + elif present < declared: + # A frame running past the bytes supplied is data stopping early, not + # a declaration disagreeing with data that is all present. A caller + # reading from a source still arriving needs to know whether waiting + # for more bytes would help, and those are different answers. + return _failed(ParseStatus.UNEXPECTED_END) + + payload = data[at:at + declared] + at += declared + signature = data[at:at + SIGNATURE_LENGTH] + at += SIGNATURE_LENGTH + + if exact and at != total: + # Unreachable while the count check above holds, and kept so a future + # change to that arithmetic cannot silently start accepting trailing + # bytes. + return _failed(ParseStatus.MALFORMED_ENVELOPE) + + return ParseResult( + True, + Owid._create( + version=version, + domain=domain, + date=date, + payload=payload, + signature=signature, + ), + ParseStatus.PARSED, + at, + ) diff --git a/owid/status.py b/owid/status.py new file mode 100644 index 0000000..df18916 --- /dev/null +++ b/owid/status.py @@ -0,0 +1,133 @@ +# **************************************************************************** +# Copyright 2026 51 Degrees Mobile Experts Limited (51degrees.com) +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# **************************************************************************** + +"""Why reading an OWID succeeded or failed.""" + +from enum import Enum + + +class ParseStatus(Enum): + """Why a parse of external data succeeded or failed. + + Malformed data arriving from outside is expected, not exceptional. An + OWID is read from whatever a caller was given, which on a public endpoint + means anything at all, so every one of these outcomes is a normal result + rather than a fault. Raising for them costs the construction and unwinding + of an exception per bad input, which is a cost whoever is sending the data + chooses the size of. + + These names are the cross-language vocabulary. Each implementation spells + its surface in its own idiom, but the set of facts reported is the same + everywhere, so a failure means the same thing whichever language read the + bytes. + """ + + #: The bytes form a structurally valid OWID. This says nothing about the + #: signature, which is a separate question answered separately. + PARSED = "Parsed" + + #: Nothing was supplied to parse. + MISSING_INPUT = "MissingInput" + + #: The input was supplied in a form this surface cannot read. + INVALID_INPUT_TYPE = "InvalidInputType" + + #: The string is not valid base 64, so there are no bytes to read. + INVALID_BASE64 = "InvalidBase64" + + #: The first byte names a version this implementation does not know. + UNSUPPORTED_VERSION = "UnsupportedVersion" + + #: The data stopped in the middle of a field. Distinct from + #: BYTE_COUNT_MISMATCH, which is a declaration that disagrees with data + #: that is all present. + UNEXPECTED_END = "UnexpectedEnd" + + #: The creator domain is not terminated, or is longer than the published + #: maximum. + INVALID_DOMAIN_ENCODING = "InvalidDomainEncoding" + + #: The declared payload byte count disagrees with the bytes actually + #: present. Checked before anything is sized by the declaration, so a + #: sender cannot make a reader allocate by claiming a large payload it did + #: not send. + BYTE_COUNT_MISMATCH = "ByteCountMismatch" + + #: The envelope is structurally consistent but larger than this runtime + #: can hold. Not a fault in the data, and deliberately distinct from the + #: data being wrong, because the same bytes may be readable elsewhere. + IMPLEMENTATION_CAPACITY_EXCEEDED = "ImplementationCapacityExceeded" + + #: The version 0 marker, which stands for an absent node inside a stream. + #: It is not an OWID and never produces one, because it carries no + #: signature and so can never verify. A framed read reports it and moves + #: past its one byte, so a caller walking a run of frames can tell an + #: absent node from a malformed one, which is the distinction the marker + #: exists for. Reported by the whole buffer read too, because the byte + #: means the same thing wherever it appears; calling it an unsupported + #: version was inaccurate, since version 0 is supported and meaningful. + ABSENT_NODE = "AbsentNode" + + #: Malformed in a way none of the above describes. A fallback for the + #: genuinely unclassified, not a substitute for naming a failure that is + #: already understood. + MALFORMED_ENVELOPE = "MalformedEnvelope" + + +class SignatureStatus(Enum): + """The outcome of asking whether an OWID's signature is genuine. + + Only two of these say anything about the signature itself. The rest say + the question could not be answered, which is a different thing and must + never be reported as a forgery. A key that cannot be fetched, a key that + cannot be decoded, or a provider that fails leaves the signature unjudged, + and a caller acting on "invalid" would reject good identifiers during an + outage. + + On 30 August 2026 the key endpoints published PEM wrapped at 76 + characters, which a strict parser rejects, and every offline verification + against them failed. The keys were fine and the identifiers were fine. + Reported as INVALID_KEY that reads as the operational fault it was; + reported as SIGNATURE_INVALID it would have read as an attack. + """ + + #: The signature is genuine for this data and this key. + SIGNATURE_VALID = "SignatureValid" + + #: The signature is well formed and does not match. The only status that + #: means the identifier should be distrusted. + SIGNATURE_INVALID = "SignatureInvalid" + + #: A signature field of the wrong length reached a verification surface + #: directly. Truncation in raw external input is a parse UNEXPECTED_END + #: instead, because there the envelope never formed. + INVALID_SIGNATURE_LENGTH = "InvalidSignatureLength" + + #: No key could be obtained, or none covers the identifier's date. The + #: signature was never examined. + KEY_UNAVAILABLE = "KeyUnavailable" + + #: Key material arrived but cannot be decoded, imported or used as the + #: required type. The fault is in the key, not the identifier. + INVALID_KEY = "InvalidKey" + + #: The work required exceeds what this runtime can hold. + IMPLEMENTATION_CAPACITY_EXCEEDED = "ImplementationCapacityExceeded" + + #: The check could not be completed for a reason that is not the + #: identifier's fault, such as a malformed key list or a cryptographic + #: provider failing on valid inputs. + VERIFICATION_ERROR = "VerificationError" diff --git a/tests/test_creator.py b/tests/test_creator.py index 2bb69aa..1c758ad 100644 --- a/tests/test_creator.py +++ b/tests/test_creator.py @@ -26,7 +26,7 @@ class CreatorTests(unittest.TestCase): def test_sign_sets_domain_date_and_version(self) -> None: crypto = Crypto.new() creator = Creator("example.com", crypto) - owid = creator.sign_string("Hello World") + owid = creator.create_string("Hello World") self.assertEqual(owid.domain, "example.com") self.assertEqual(owid.version, Version.VERSION3) self.assertEqual(len(owid.signature), 64) @@ -35,8 +35,8 @@ def test_sign_sets_domain_date_and_version(self) -> None: def test_sign_string_and_sign_bytes_match(self) -> None: crypto = Crypto.new() creator = Creator("example.com", crypto) - from_string = creator.sign_string("payload") - from_bytes = creator.sign_bytes(b"payload") + from_string = creator.create_string("payload") + from_bytes = creator.create(b"payload") self.assertEqual(from_string.payload, from_bytes.payload) def test_empty_domain_raises(self) -> None: @@ -61,7 +61,7 @@ def test_from_configuration(self) -> None: ) creator = Creator.from_configuration(configuration) self.assertEqual(creator.domain, "example.com") - owid = creator.sign_string("Hello World") + owid = creator.create_string("Hello World") # The OWID must verify with the public key from the original crypto. self.assertTrue(owid.verify_with_public_key(crypto.public_key_pem(), [])) diff --git a/tests/test_domain_length.py b/tests/test_domain_length.py index 9b5ea57..0569414 100644 --- a/tests/test_domain_length.py +++ b/tests/test_domain_length.py @@ -90,12 +90,12 @@ def test_maximum_length_domain_parses(self) -> None: trips through the writer and the reader unchanged.""" self.assertEqual(len(MAXIMUM_DOMAIN), MAXIMUM_DOMAIN_LENGTH) - owid = Owid.from_byte_array(envelope(terminated(MAXIMUM_DOMAIN))) + owid = Owid._from_byte_array_or_raise(envelope(terminated(MAXIMUM_DOMAIN))) self.assertEqual(owid.domain, MAXIMUM_DOMAIN) self.assertEqual(owid.payload, PAYLOAD) self.assertEqual(owid.signature, SIGNATURE) - again = Owid.from_byte_array(owid.as_byte_array()) + again = Owid._from_byte_array_or_raise(owid.as_byte_array()) self.assertEqual(again.domain, MAXIMUM_DOMAIN) self.assertEqual(again, owid) @@ -107,7 +107,7 @@ def test_one_character_over_the_maximum_is_refused(self) -> None: self.assertEqual(len(domain), MAXIMUM_DOMAIN_LENGTH + 1) with self.assertRaises(OwidError) as raised: - Owid.from_byte_array(envelope(terminated(domain))) + Owid._from_byte_array_or_raise(envelope(terminated(domain))) self.assertIn( "'{0}'".format(MAXIMUM_DOMAIN_LENGTH), str(raised.exception) @@ -127,7 +127,7 @@ def test_missing_terminator_is_refused_within_the_bound(self) -> None: started = time.perf_counter() for _ in range(1000): with self.assertRaises(OwidError): - Owid.from_byte_array(raw) + Owid._from_byte_array_or_raise(raw) elapsed = time.perf_counter() - started self.assertLess( @@ -140,7 +140,7 @@ def test_missing_terminator_is_refused_within_the_bound(self) -> None: tracemalloc.start() try: with self.assertRaises(OwidError): - Owid.from_byte_array(raw) + Owid._from_byte_array_or_raise(raw) _, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() @@ -164,7 +164,7 @@ def test_distant_terminator_does_not_size_the_domain(self) -> None: tracemalloc.start() try: with self.assertRaises(OwidError): - Owid.from_byte_array(raw) + Owid._from_byte_array_or_raise(raw) _, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() @@ -180,9 +180,9 @@ def test_library_output_parses(self) -> None: still parses and still verifies, so the bound refuses nothing the library itself produces.""" crypto = Crypto.new() - original = Creator(MAXIMUM_DOMAIN, crypto).sign_bytes(PAYLOAD) + original = Creator(MAXIMUM_DOMAIN, crypto).create(PAYLOAD) - parsed = Owid.from_byte_array(original.as_byte_array()) + parsed = Owid._from_byte_array_or_raise(original.as_byte_array()) self.assertEqual(parsed.domain, MAXIMUM_DOMAIN) self.assertEqual(parsed, original) @@ -215,14 +215,14 @@ def test_maximum_length_domain_is_written_and_parses_back(self) -> None: """A domain of exactly the maximum length is written and the value parses back unchanged, so the bound refuses nothing at or under the maximum.""" - owid = Owid( + owid = Owid._create( version=Version.VERSION3, domain=MAXIMUM_DOMAIN, payload=PAYLOAD, signature=SIGNATURE, ) - parsed = Owid.from_byte_array(owid.as_byte_array()) + parsed = Owid._from_byte_array_or_raise(owid.as_byte_array()) self.assertEqual(parsed.domain, MAXIMUM_DOMAIN) self.assertEqual(parsed.payload, PAYLOAD) @@ -257,7 +257,7 @@ def test_writing_a_domain_over_the_maximum_is_refused(self) -> None: on the OWID directly, is refused when it is serialised, so the library never emits an OWID it would refuse to read.""" domain = MAXIMUM_DOMAIN + "c" - owid = Owid( + owid = Owid._create( version=Version.VERSION3, domain=domain, payload=PAYLOAD, @@ -280,7 +280,7 @@ def test_refusal_comes_before_any_signature_is_computed(self) -> None: refused without the counter moving again.""" counting = CountingCrypto(Crypto.new()) - Creator(MAXIMUM_DOMAIN, counting).sign_bytes(PAYLOAD) + Creator(MAXIMUM_DOMAIN, counting).create(PAYLOAD) self.assertEqual(counting.sign_calls, 1) with self.assertRaises(OwidError): diff --git a/tests/test_owid.py b/tests/test_owid.py index 3428d95..2db85c6 100644 --- a/tests/test_owid.py +++ b/tests/test_owid.py @@ -19,11 +19,18 @@ import unittest +from owid.io import SIGNATURE_LENGTH from owid import Crypto, Owid, Version from tests import fixtures +def _parsed(result): + """The OWID from a result a test has already decided is valid.""" + assert result.ok, "expected valid input, got {0}".format(result.status) + return result.owid + + class CanonicalWireVectorTests(unittest.TestCase): """The three canonical vectors must round trip byte exact. @@ -34,7 +41,7 @@ class CanonicalWireVectorTests(unittest.TestCase): def _assert_round_trip(self, vector: str) -> Owid: original = fixtures.decode_unpadded(vector) - owid = Owid.from_byte_array(original) + owid = Owid._from_byte_array_or_raise(original) self.assertEqual( owid.as_byte_array(), original, @@ -83,11 +90,11 @@ def test_simple_and_utf8_verify(self) -> None: for name, data in fixtures.ALL_LANGUAGES: spki = data["spki"] with self.subTest(language=name, fixture="simple"): - simple = Owid.from_base64(data["simple"]) + simple = _parsed(Owid.parse(data["simple"])) self.assertEqual(simple.payload_as_string(), "example") self.assertTrue(simple.verify_with_public_key(spki, [])) with self.subTest(language=name, fixture="utf8"): - utf8 = Owid.from_base64(data["utf8"]) + utf8 = _parsed(Owid.parse(data["utf8"])) self.assertEqual( utf8.payload_as_string(), fixtures.UTF8_PAYLOAD, @@ -98,8 +105,8 @@ def test_simple_and_utf8_verify(self) -> None: def test_chain_verifies_with_root_as_other(self) -> None: for name, data in fixtures.ALL_LANGUAGES: spki = data["spki"] - root = Owid.from_base64(data["chain_root"]) - party = Owid.from_base64(data["chain_party"]) + root = _parsed(Owid.parse(data["chain_root"])) + party = _parsed(Owid.parse(data["chain_party"])) with self.subTest(language=name): self.assertEqual(root.payload_as_string(), "root") self.assertEqual(party.payload_as_string(), "party") @@ -111,7 +118,7 @@ def test_chain_verifies_with_root_as_other(self) -> None: def test_chain_party_fails_without_others(self) -> None: for name, data in fixtures.ALL_LANGUAGES: spki = data["spki"] - party = Owid.from_base64(data["chain_party"]) + party = _parsed(Owid.parse(data["chain_party"])) with self.subTest(language=name): self.assertFalse( party.verify_with_public_key(spki, []), @@ -121,19 +128,19 @@ def test_chain_party_fails_without_others(self) -> None: def test_tampered_fixtures_fail(self) -> None: for name, data in fixtures.ALL_LANGUAGES: spki = data["spki"] - root = Owid.from_base64(data["chain_root"]) + root = _parsed(Owid.parse(data["chain_root"])) for fixture in ("simple", "utf8", "chain_root"): with self.subTest(language=name, fixture=fixture): - tampered = Owid.from_base64( - fixtures.flip_last_byte(data[fixture]) + tampered = _parsed(Owid.parse( + fixtures.flip_last_byte(data[fixture])) ) self.assertFalse( tampered.verify_with_public_key(spki, []), "a flipped signature byte must fail to verify", ) with self.subTest(language=name, fixture="chain_party"): - tampered_party = Owid.from_base64( - fixtures.flip_last_byte(data["chain_party"]) + tampered_party = _parsed(Owid.parse( + fixtures.flip_last_byte(data["chain_party"])) ) self.assertFalse( tampered_party.verify_with_public_key(spki, [root]), @@ -149,12 +156,12 @@ def test_sign_and_self_verify(self) -> None: crypto = Crypto.new() creator = Creator("example.com", crypto) - owid = creator.sign_string("Hello World") + owid = creator.create_string("Hello World") # Verifies with the crypto instance and with the exported public key. self.assertTrue(owid.verify_with_crypto(crypto, [])) self.assertTrue(owid.verify_with_public_key(crypto.public_key_pem(), [])) # A copy decoded from base 64 also verifies. - copy = Owid.from_base64(owid.as_base64()) + copy = _parsed(Owid.parse(owid.as_base64())) self.assertEqual(copy, owid) self.assertTrue(copy.verify_with_crypto(crypto, [])) @@ -163,19 +170,29 @@ def test_tampered_self_signed_fails(self) -> None: crypto = Crypto.new() creator = Creator("example.com", crypto) - owid = creator.sign_string("Hello World") - # Change a payload byte and verification must fail. - owid.payload = b"Hello Worle" - self.assertFalse(owid.verify_with_crypto(crypto, [])) + owid = creator.create_string("Hello World") + # Tampering happens to the bytes in transit, not to an object already + # in memory, and it can no longer be done in memory because the fields + # are read only. So the envelope is serialised, a payload byte is + # changed, and the result is read back the way a receiver would read + # it. + raw = bytearray(owid.as_byte_array()) + at = len(raw) - SIGNATURE_LENGTH - len(owid.payload) + raw[at] ^= 0xFF + result = Owid.parse_bytes(bytes(raw)) + self.assertTrue(result.ok, result.status) + self.assertFalse(result.owid.verify_with_crypto(crypto, [])) def test_sign_with_others_round_trip(self) -> None: from owid import Creator crypto = Crypto.new() creator = Creator("example.com", crypto) - root = creator.sign_string("root") - party = Owid(payload=b"party") - creator.sign_with_others(party, [root]) + root = creator.create_string("root") + # Created with the root as the other it is signed alongside, rather + # than assembled and then signed, because a caller no longer makes an + # OWID and hands it over to have a signature put on it. + party = creator.create(b"party", [root]) # Party verifies with the root as the single other, and fails alone. self.assertTrue(party.verify_with_crypto(crypto, [root])) self.assertFalse(party.verify_with_crypto(crypto, [])) @@ -185,10 +202,10 @@ class PayloadAndSerializationTests(unittest.TestCase): """Payload accessors and the empty OWID marker.""" def test_payload_accessors(self) -> None: - owid = Owid(payload=bytes([0x01, 0x03])) + owid = Owid._create(payload=bytes([0x01, 0x03])) self.assertEqual(owid.payload_as_printable(), "0103") self.assertEqual(owid.payload_as_base64(), "AQM=") - self.assertEqual(Owid(payload=b"example").payload_as_string(), "example") + self.assertEqual(Owid._create(payload=b"example").payload_as_string(), "example") def test_unpadded_and_padded_decode(self) -> None: # The supplier vector is unpadded. Adding padding must decode to the @@ -196,21 +213,21 @@ def test_unpadded_and_padded_decode(self) -> None: unpadded = fixtures.SUPPLIER_VECTOR padded = unpadded + "=" self.assertEqual( - Owid.from_base64(unpadded), Owid.from_base64(padded) + _parsed(Owid.parse(unpadded)), _parsed(Owid.parse(padded)) ) def test_empty_marker(self) -> None: buffer = bytearray() Owid.empty_to_buffer(buffer) self.assertEqual(bytes(buffer), bytes([0])) - owid = Owid.from_byte_array(bytes(buffer)) + owid = Owid._from_byte_array_or_raise(bytes(buffer)) self.assertEqual(owid.version, Version.EMPTY) def test_short_buffer_raises(self) -> None: from owid import OwidError with self.assertRaises(OwidError): - Owid.from_byte_array(bytes([0x03, 0x61])) + Owid._from_byte_array_or_raise(bytes([0x03, 0x61])) if __name__ == "__main__": diff --git a/tests/test_parse_contract.py b/tests/test_parse_contract.py new file mode 100644 index 0000000..fb645a4 --- /dev/null +++ b/tests/test_parse_contract.py @@ -0,0 +1,399 @@ +# **************************************************************************** +# Copyright 2026 51 Degrees Mobile Experts Limited (51degrees.com) +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# **************************************************************************** + +"""What the parse and creation surfaces promise, tested as a contract. + +Two promises are kept here. The first is that reading external data always +answers rather than raising, and answers with the same three facts: whether it +worked, the value only when it did, and a named reason either way. The second +is that an OWID can only arrive by parsing one or by a creator signing one, so +no caller can hold a half made or altered identifier. +""" + +from __future__ import annotations + +import unittest + +from owid import Creator, Crypto, Owid +from owid.error import OwidError +from owid.io import SIGNATURE_LENGTH +from owid.status import ParseStatus, SignatureStatus + + +def _creator() -> Creator: + return Creator("example.com", Crypto.new()) + + +class ParseContractTests(unittest.TestCase): + """The three facts a parse always reports.""" + + def test_success_reports_ok_a_value_and_parsed(self) -> None: + owid = _creator().create(b"\x01\x02\x03") + + result = Owid.parse_bytes(owid.as_byte_array()) + + self.assertTrue(result.ok) + self.assertIsNotNone(result.owid) + self.assertEqual(ParseStatus.PARSED, result.status) + self.assertTrue(result, "the result is truthy on success") + + def test_empty_payload_parses(self) -> None: + """Having nothing to say is allowed: the payload is what the creator + had to say, and an OWID carrying nothing is still an OWID.""" + owid = _creator().create(b"") + + result = Owid.parse_bytes(owid.as_byte_array()) + + self.assertTrue(result.ok, result.status) + self.assertEqual(b"", result.owid.payload) + + def test_large_payload_parses(self) -> None: + """A megabyte parses. The format's limit is the wire format's, and how + much an application will accept is that application's policy rather + than something this library decides for it.""" + payload = bytes(1024 * 1024) + owid = _creator().create(payload) + + result = Owid.parse_bytes(owid.as_byte_array()) + + self.assertTrue(result.ok, result.status) + self.assertEqual(len(payload), len(result.owid.payload)) + + def test_absent_input_is_missing_input(self) -> None: + for value in (None, ""): + with self.subTest(value=value): + result = Owid.parse(value) + self.assertFalse(result.ok) + self.assertIsNone(result.owid) + self.assertEqual(ParseStatus.MISSING_INPUT, result.status) + + def test_none_buffer_is_missing_input(self) -> None: + result = Owid.parse_bytes(None) + + self.assertFalse(result.ok) + self.assertIsNone(result.owid) + self.assertEqual(ParseStatus.MISSING_INPUT, result.status) + + def test_invalid_base64_is_reported_not_raised(self) -> None: + result = Owid.parse("not base 64 at all!!") + + self.assertFalse(result.ok) + self.assertIsNone(result.owid) + self.assertEqual(ParseStatus.INVALID_BASE64, result.status) + + def test_unsupported_version_is_reported(self) -> None: + raw = bytearray(_creator().create(b"x").as_byte_array()) + raw[0] = 9 + + result = Owid.parse_bytes(bytes(raw)) + + self.assertFalse(result.ok) + self.assertEqual(ParseStatus.UNSUPPORTED_VERSION, result.status) + + def test_trailing_byte_is_refused(self) -> None: + raw = _creator().create(b"x").as_byte_array() + b"\x00" + + result = Owid.parse_bytes(raw) + + self.assertFalse(result.ok) + self.assertEqual(ParseStatus.BYTE_COUNT_MISMATCH, result.status) + + +class ConstructionBoundaryTests(unittest.TestCase): + """An OWID arrives from a parse or a creator, and from nowhere else.""" + + def test_direct_construction_is_refused(self) -> None: + """Python cannot make a constructor package private, so the boundary + is kept by refusing a caller who has not come through one of the two + allowed paths. Without it an unsigned OWID could be handed to code + that cannot tell the difference.""" + with self.assertRaises(OwidError) as caught: + Owid() + + self.assertIn("cannot be constructed directly", str(caught.exception)) + + def test_state_cannot_be_rebound(self) -> None: + owid = _creator().create(b"abc") + + for field in ("version", "domain", "date", "payload", "signature"): + with self.subTest(field=field): + with self.assertRaises(AttributeError): + setattr(owid, field, None) + + def test_payload_is_immutable(self) -> None: + """bytes rather than a mutable buffer, so what a caller was given + cannot be written into.""" + owid = _creator().create(b"abc") + + self.assertIsInstance(owid.payload, bytes) + self.assertIsInstance(owid.signature, bytes) + + +class ParsingIsNotVerificationTests(unittest.TestCase): + """Two separate questions with two separate answers.""" + + def test_structurally_valid_but_unsigned_parses_then_fails(self) -> None: + crypto = Crypto.new() + owid = Creator("example.com", crypto).create(b"\x04\x05\x06") + raw = bytearray(owid.as_byte_array()) + raw[-1] ^= 0xFF + + result = Owid.parse_bytes(bytes(raw)) + + self.assertTrue( + result.ok, + "flipping a signature byte leaves the envelope readable") + self.assertEqual(ParseStatus.PARSED, result.status) + self.assertFalse( + result.owid.verify_with_crypto(crypto, []), + "and the signature is then found not to match") + + def test_no_verification_happens_during_a_failed_parse(self) -> None: + """A malformed identifier must be refused before anything reaches a + key or a signature check.""" + raw = bytearray(_creator().create(b"x").as_byte_array()) + raw[0] = 9 + + result = Owid.parse_bytes(bytes(raw)) + + self.assertFalse(result.ok) + self.assertIsNone( + result.owid, + "no value means nothing exists on which to check a signature") + + +if __name__ == "__main__": + unittest.main() + + +class EveryFailureConditionTests(unittest.TestCase): + """One test per failure the vocabulary can report. + + James Rosewell asked that every failure condition is covered. Where a + member cannot be reached in Python the reason is stated here and on the + member itself, rather than a path being invented to reach it. + """ + + def test_missing_input(self) -> None: + self.assertEqual( + ParseStatus.MISSING_INPUT, Owid.parse_bytes(b"").status) + + def test_invalid_input_type(self) -> None: + """Python does not check types at the boundary, so something that is + neither text nor bytes reaches the reader and is named rather than + raising a TypeError from somewhere deeper.""" + self.assertEqual( + ParseStatus.INVALID_INPUT_TYPE, Owid.parse(12345).status) + self.assertEqual( + ParseStatus.INVALID_INPUT_TYPE, Owid.parse_bytes(12345).status) + + def test_invalid_base64(self) -> None: + self.assertEqual( + ParseStatus.INVALID_BASE64, Owid.parse("not base 64!!").status) + + def test_unsupported_version(self) -> None: + raw = bytearray(_creator().create(b"x").as_byte_array()) + raw[0] = 9 + self.assertEqual( + ParseStatus.UNSUPPORTED_VERSION, Owid.parse_bytes(bytes(raw)).status) + + def test_empty_marker_is_an_absent_node(self) -> None: + """The version 0 marker stands for an absent node inside a stream. It + has no signature, so it can never verify, and no value is handed back. It + is named for what it is rather than called an unsupported version, + because version 0 is supported and meaningful, it simply is not an + OWID.""" + self.assertEqual( + ParseStatus.ABSENT_NODE, Owid.parse_bytes(b"\x00").status) + + def test_unexpected_end(self) -> None: + """Data that stops inside a field, before the declared length is even + read. Distinct from a declaration disagreeing with data that is here.""" + raw = _creator().create(b"x").as_byte_array()[:3] + self.assertEqual( + ParseStatus.UNEXPECTED_END, Owid.parse_bytes(raw).status) + + def test_invalid_domain_encoding(self) -> None: + """A domain that never terminates within the published maximum.""" + raw = bytes([3]) + b"a" * 300 + self.assertEqual( + ParseStatus.INVALID_DOMAIN_ENCODING, Owid.parse_bytes(raw).status) + + def test_byte_count_mismatch_when_longer(self) -> None: + raw = _creator().create(b"x").as_byte_array() + b"\x00" + self.assertEqual( + ParseStatus.BYTE_COUNT_MISMATCH, Owid.parse_bytes(raw).status) + + def test_byte_count_mismatch_when_signature_is_short(self) -> None: + """The declared payload cannot leave exactly the signature the version + requires, which is the finding whichever way the bytes fall short.""" + raw = _creator().create(b"x").as_byte_array()[:-1] + self.assertEqual( + ParseStatus.BYTE_COUNT_MISMATCH, Owid.parse_bytes(raw).status) + + def test_implementation_capacity_and_malformed_are_unreachable(self) -> None: + """IMPLEMENTATION_CAPACITY_EXCEEDED cannot be reached in Python: + integers are unbounded and a declaration large enough to matter can + never agree with the bytes actually present, so the count check refuses + it first. MALFORMED_ENVELOPE is likewise unreachable while the count + check holds, and both are kept as backstops so that a future change to + that arithmetic cannot pass silently. + + This test exists to record that, so the gap is a stated decision rather + than something nobody noticed.""" + self.assertIn(ParseStatus.IMPLEMENTATION_CAPACITY_EXCEEDED, ParseStatus) + self.assertIn(ParseStatus.MALFORMED_ENVELOPE, ParseStatus) + + +class SignatureStatusTests(unittest.TestCase): + """Keeping "could not check" apart from "does not match".""" + + def test_valid(self) -> None: + crypto = Crypto.new() + owid = Creator("example.com", crypto).create(b"abc") + self.assertEqual( + SignatureStatus.SIGNATURE_VALID, + owid.signature_status(crypto.public_key_pem(), [])) + + def test_invalid_only_when_it_really_does_not_match(self) -> None: + crypto = Crypto.new() + owid = Creator("example.com", crypto).create(b"abc") + raw = bytearray(owid.as_byte_array()) + raw[-1] ^= 0xFF + tampered = Owid.parse_bytes(bytes(raw)).owid + self.assertEqual( + SignatureStatus.SIGNATURE_INVALID, + tampered.signature_status(crypto.public_key_pem(), [])) + + def test_no_key_is_not_a_forgery(self) -> None: + crypto = Crypto.new() + owid = Creator("example.com", crypto).create(b"abc") + self.assertEqual( + SignatureStatus.KEY_UNAVAILABLE, owid.signature_status("", [])) + + def test_unreadable_key_is_not_a_forgery(self) -> None: + """This is the case that happened. The key endpoints served PEM a + strict parser rejects, and reporting it as a forgery would have read as + an attack rather than the outage it was.""" + crypto = Crypto.new() + owid = Creator("example.com", crypto).create(b"abc") + self.assertEqual( + SignatureStatus.INVALID_KEY, + owid.signature_status( + "-----BEGIN PUBLIC KEY-----\nnot base 64\n" + "-----END PUBLIC KEY-----", [])) + + def test_remaining_members_are_unreachable_here(self) -> None: + """INVALID_SIGNATURE_LENGTH cannot be reached from a parsed OWID, + because a parse only succeeds when the signature is exactly the + required length; the guard covers an OWID reaching this by another + route. VERIFICATION_ERROR needs the cryptographic provider to fail on + inputs that are themselves fine, and IMPLEMENTATION_CAPACITY_EXCEEDED + needs more data than this runtime can hold. Both are kept because + neither may ever be reported as a forgery.""" + self.assertIn(SignatureStatus.INVALID_SIGNATURE_LENGTH, SignatureStatus) + self.assertIn(SignatureStatus.VERIFICATION_ERROR, SignatureStatus) + self.assertIn( + SignatureStatus.IMPLEMENTATION_CAPACITY_EXCEEDED, SignatureStatus) + + + def test_an_absent_node_is_skipped_and_the_next_frame_read(self) -> None: + """The distinction the marker exists for: a caller walking a run of + frames can tell an absent node from a malformed one, and carry on.""" + envelope = _creator().create(b"after the gap").as_byte_array() + data = b"\x00" + envelope + + first = Owid.parse_prefix(data) + self.assertFalse(first.ok, "a marker is not an OWID") + self.assertIsNone(first.owid) + self.assertEqual(ParseStatus.ABSENT_NODE, first.status) + self.assertEqual(1, first.consumed, "and it moves past the one byte") + + second = Owid.parse_prefix(data[first.consumed:]) + self.assertTrue(second.ok, second.status) + self.assertEqual(b"after the gap", second.owid.payload) + + +class FramedReadTests(unittest.TestCase): + """Reading one envelope from a buffer that holds more after it. + + The two reads differ in exactly one place. A whole buffer knows where the + envelope ends, so the declared payload must leave exactly the signature. A + framed read does not, because what follows may be the next envelope rather + than rubbish. + """ + + def test_walks_a_run_of_envelopes(self) -> None: + creator = _creator() + first = creator.create(b"first").as_byte_array() + second = creator.create(b"second").as_byte_array() + data = first + second + + payloads = [] + while data: + result = Owid.parse_prefix(data) + self.assertTrue(result.ok, result.status) + payloads.append(result.owid.payload) + data = data[result.consumed:] + + self.assertEqual([b"first", b"second"], payloads) + + def test_the_whole_buffer_read_refuses_the_same_bytes(self) -> None: + """Where nothing else could own the trailing bytes, they are a + disagreement rather than the next envelope.""" + creator = _creator() + data = (creator.create(b"first").as_byte_array() + + creator.create(b"second").as_byte_array()) + + result = Owid.parse_bytes(data) + + self.assertFalse(result.ok) + self.assertEqual(ParseStatus.BYTE_COUNT_MISMATCH, result.status) + + def test_a_truncated_envelope_is_refused_and_consumes_nothing(self) -> None: + raw = _creator().create(b"payload").as_byte_array()[:-1] + + result = Owid.parse_prefix(raw) + + self.assertFalse(result.ok) + # Data stopping early, not a declaration disagreeing with data that is + # all present. A caller reading from a source still arriving needs to + # know whether waiting for more bytes would help. + self.assertEqual(ParseStatus.UNEXPECTED_END, result.status) + self.assertEqual(0, result.consumed) + # Reading it again gives the same answer, since nothing moved. + self.assertEqual(result.status, Owid.parse_prefix(raw).status) + + def test_the_framed_read_reports_the_same_reasons(self) -> None: + """Everything except what follows the envelope is judged identically, + so a caller does not have to learn two vocabularies.""" + good = _creator().create(b"payload").as_byte_array() + unknown = bytearray(good) + unknown[0] = 9 + + for name, raw, expected in ( + ("stops inside a field", good[:3], ParseStatus.UNEXPECTED_END), + ("nothing supplied", b"", ParseStatus.MISSING_INPUT), + ("unknown version", bytes(unknown), + ParseStatus.UNSUPPORTED_VERSION), + ("the absent marker", b"\x00", + ParseStatus.ABSENT_NODE), + ): + with self.subTest(name): + self.assertEqual( + expected, Owid.parse_prefix(raw).status) + self.assertEqual( + expected, Owid.parse_bytes(raw).status) diff --git a/tests/test_payload_length.py b/tests/test_payload_length.py index 676e8c7..576b5ef 100644 --- a/tests/test_payload_length.py +++ b/tests/test_payload_length.py @@ -58,7 +58,7 @@ class PayloadLengthTests(unittest.TestCase): def test_declared_length_matches_parses(self) -> None: """The declared length matches the bytes present, the signature is the last 64 bytes, and the envelope parses to the same payload.""" - owid = Owid.from_byte_array( + owid = Owid._from_byte_array_or_raise( envelope(len(PAYLOAD), PAYLOAD, SIGNATURE) ) self.assertEqual(owid.payload, PAYLOAD) @@ -69,7 +69,7 @@ def test_matching_one_mebibyte_payload_parses(self) -> None: """A matching large payload is valid; size policy belongs upstream.""" payload = b"\x5a" * (1024 * 1024) - owid = Owid.from_byte_array( + owid = Owid._from_byte_array_or_raise( envelope(len(payload), payload, SIGNATURE) ) @@ -80,8 +80,8 @@ def test_library_output_parses(self) -> None: still parses, so the check agrees with what the library itself produces.""" crypto = Crypto.new() - original = Creator(DOMAIN, crypto).sign_bytes(PAYLOAD) - parsed = Owid.from_byte_array(original.as_byte_array()) + original = Creator(DOMAIN, crypto).create(PAYLOAD) + parsed = Owid._from_byte_array_or_raise(original.as_byte_array()) self.assertEqual(parsed.payload, PAYLOAD) self.assertEqual(parsed, original) self.assertTrue(parsed.verify_with_crypto(crypto, [])) @@ -95,7 +95,7 @@ def test_declared_length_off_by_one_is_refused(self) -> None: for declared in (len(PAYLOAD) - 1, len(PAYLOAD) + 1): with self.subTest(declared=declared): with self.assertRaises(OwidError) as raised: - Owid.from_byte_array( + Owid._from_byte_array_or_raise( envelope(declared, PAYLOAD, SIGNATURE) ) message = str(raised.exception) @@ -107,14 +107,14 @@ def test_trailing_byte_after_signature_is_refused(self) -> None: must be the end of the envelope.""" longer = envelope(len(PAYLOAD), PAYLOAD, SIGNATURE) + b"\x00" with self.assertRaises(OwidError): - Owid.from_byte_array(longer) + Owid._from_byte_array_or_raise(longer) def test_short_signature_is_refused(self) -> None: """A short signature is refused. The declared payload length is right for the payload, but the bytes after it are fewer than a signature.""" with self.assertRaises(OwidError): - Owid.from_byte_array( + Owid._from_byte_array_or_raise( envelope(len(PAYLOAD), PAYLOAD, SIGNATURE[:-1]) ) @@ -137,7 +137,7 @@ def test_mismatched_large_declaration_is_refused_without_allocating( started = time.perf_counter() for _ in range(1000): with self.assertRaises(OwidError): - Owid.from_byte_array(raw) + Owid._from_byte_array_or_raise(raw) elapsed = time.perf_counter() - started self.assertLess( elapsed, @@ -149,7 +149,7 @@ def test_mismatched_large_declaration_is_refused_without_allocating( tracemalloc.start() try: with self.assertRaises(OwidError): - Owid.from_byte_array(raw) + Owid._from_byte_array_or_raise(raw) _, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() @@ -164,7 +164,7 @@ def test_mismatched_large_declaration_is_refused_without_allocating( def test_empty_payload_parses(self) -> None: """A declared length of zero followed by the 64 byte signature parses, so the check does not refuse the smallest valid payload.""" - owid = Owid.from_byte_array(envelope(0, b"", SIGNATURE)) + owid = Owid._from_byte_array_or_raise(envelope(0, b"", SIGNATURE)) self.assertEqual(owid.payload, b"") self.assertEqual(owid.signature, SIGNATURE) diff --git a/tests/test_readme.py b/tests/test_readme.py new file mode 100644 index 0000000..8bf0c65 --- /dev/null +++ b/tests/test_readme.py @@ -0,0 +1,73 @@ +# **************************************************************************** +# Copyright 2026 51 Degrees Mobile Experts Limited (51degrees.com) +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# **************************************************************************** + +"""Runs the Python examples in the README so that documentation naming a +method that does not exist fails the build. + +The examples are taken in the order they appear and run in one namespace, +because the later ones use the creator and the OWID the first one makes, which +is how a reader follows them. Checks are appended so that the examples are +shown to do what the surrounding text says they do rather than merely to run. +""" + +from __future__ import annotations + +import os +import re +import unittest + +_HERE = os.path.dirname(os.path.abspath(__file__)) +README = os.path.join(os.path.dirname(_HERE), "README.md") + +#: Fenced blocks marked as Python, in the order they appear. +_BLOCK = re.compile(r"```python\r?\n(.*?)```", re.DOTALL) + + +def _examples(): + with open(README, "r", encoding="utf-8") as handle: + return _BLOCK.findall(handle.read()) + + +class ReadmeTests(unittest.TestCase): + """Every documented example runs, and does what the text says it does.""" + + def test_readme_still_carries_its_examples(self) -> None: + """A README that lost its examples would pass the run below without + having run anything, so the count is asserted first.""" + self.assertGreaterEqual(len(_examples()), 4) + + def test_readme_examples_run(self) -> None: + namespace = {"__name__": "readme_example"} + for index, example in enumerate(_examples()): + with self.subTest(example=index): + exec(compile(example, "README.md#{0}".format(index), "exec"), + namespace) + + # The first example read its own identifier back. + result = namespace["result"] + self.assertTrue(result.ok, result.status) + self.assertEqual(namespace["owid"], result.owid) + + # The signature status example found a genuine signature. + status = namespace["status"] + self.assertEqual("SignatureValid", status.value) + + # The framed walk stepped over the absent node and read both OWIDs. + self.assertEqual(["first", "second"], namespace["payloads"]) + + +if __name__ == "__main__": + unittest.main()