#!/usr/bin/env python3
"""Reference verifier for BMP run manifests, schema bmp.manifest/v1 (storage section 5.5, FR-DR.1).

A second implementation of the manifest's canonical form and of its signature check, written without reference to
the platform's code: RFC 8785 (JSON Canonicalization Scheme) over the manifest without its "signature" member,
checked with Ed25519 against a public key from manifest-signing-keys.json. If this tool and the platform disagree
about one byte, a manifest could be verified by one and refused by the other, and a backup whose commit marker only
BMP can read is not restorable without BMP.

Needs pyca/cryptography.

  python docs/specs/manifest/reference/manifest_ref.py verify --manifest _manifest.json --keys manifest-signing-keys.json [--json]
  python docs/specs/manifest/reference/manifest_ref.py canonicalize --manifest _manifest.json

manifest-signing-keys.json: {"keys": [{"keyId": "<uuid>", "algorithm": "Ed25519", "publicKey": "<base64>"}]}
"""

import argparse
import base64
import json
import re
import sys

MAX_BYTES = 1 << 20
MAX_SAFE_INTEGER = (1 << 53) - 1
SCHEMA = re.compile(r"^bmp\.manifest/v(?P<major>[0-9]+)(\.[0-9]+)?$")


class Refused(Exception):
    """A manifest that must not be trusted."""


def _no_duplicates(pairs):
    seen = {}
    for name, value in pairs:
        if name in seen:
            raise Refused(f"the member '{name}' appears twice in one object")
        seen[name] = value
    return seen


def _no_fractions(text):
    raise Refused(f"'{text}' is not an integer a manifest may hold")


def _no_constants(text):
    raise Refused(f"'{text}' is not JSON")


def parse(data):
    if len(data) > MAX_BYTES:
        raise Refused(f"the manifest is larger than {MAX_BYTES} bytes")
    try:
        text = data.decode("utf-8")
    except UnicodeDecodeError as error:
        raise Refused(f"the manifest is not UTF-8: {error}") from error
    try:
        return json.loads(
            text, object_pairs_hook=_no_duplicates, parse_float=_no_fractions, parse_constant=_no_constants
        )
    except json.JSONDecodeError as error:
        raise Refused(f"the manifest is not JSON: {error}") from error


def _string(value):
    out = ['"']
    for character in value:
        code = ord(character)
        if character == '"':
            out.append('\\"')
        elif character == "\\":
            out.append("\\\\")
        elif character == "\b":
            out.append("\\b")
        elif character == "\f":
            out.append("\\f")
        elif character == "\n":
            out.append("\\n")
        elif character == "\r":
            out.append("\\r")
        elif character == "\t":
            out.append("\\t")
        elif code < 0x20:
            out.append("\\u%04x" % code)
        elif 0xD800 <= code <= 0xDFFF:
            raise Refused("the manifest holds half of a character")
        else:
            out.append(character)
    out.append('"')
    return "".join(out)


def canonical(value):
    """RFC 8785: members sorted by UTF-16 code units, no whitespace, ECMAScript string and number forms."""
    if isinstance(value, dict):
        members = sorted(value.items(), key=lambda member: member[0].encode("utf-16-be"))
        return "{" + ",".join(_string(name) + ":" + canonical(item) for name, item in members) + "}"
    if isinstance(value, list):
        return "[" + ",".join(canonical(item) for item in value) + "]"
    if isinstance(value, str):
        return _string(value)
    if value is True:
        return "true"
    if value is False:
        return "false"
    if value is None:
        return "null"
    if isinstance(value, int):
        if abs(value) > MAX_SAFE_INTEGER:
            raise Refused(f"'{value}' is not an integer a manifest may hold")
        return str(value)
    raise Refused(f"the manifest holds a value this reader does not know: {value!r}")


def canonical_bytes(value):
    return canonical(value).encode("utf-8")


def verify(data, keys):
    """Returns (key id, manifest) or raises Refused."""
    from cryptography.exceptions import InvalidSignature
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

    manifest = parse(data)
    if not isinstance(manifest, dict):
        raise Refused("the manifest is not a JSON object")
    canonical_bytes(manifest)

    match = SCHEMA.match(manifest.get("schema") if isinstance(manifest.get("schema"), str) else "")
    if not match or match.group("major") != "1":
        raise Refused("the manifest is not of a schema this reader knows (bmp.manifest/v1)")

    signature = manifest.get("signature")
    if not isinstance(signature, dict):
        raise Refused("the manifest is not signed")
    if signature.get("alg") != "Ed25519" or signature.get("canonicalization") != "RFC8785":
        raise Refused("the manifest is signed in a way this reader does not know")

    key_id = signature.get("keyId")
    public_key = keys.get(key_id)
    if public_key is None:
        raise Refused(f"the manifest is signed with key {key_id}, which is not known")

    try:
        value = base64.b64decode(signature.get("value", ""), validate=True)
    except (ValueError, TypeError) as error:
        raise Refused("the manifest's signature is not base64") from error
    if len(value) != 64:
        raise Refused("the manifest's signature is not an Ed25519 signature")

    body = {name: item for name, item in manifest.items() if name != "signature"}
    try:
        Ed25519PublicKey.from_public_bytes(public_key).verify(value, canonical_bytes(body))
    except InvalidSignature as error:
        raise Refused("the manifest's signature does not match its content") from error

    return key_id, manifest


def load_keys(path):
    with open(path, "rb") as handle:
        document = json.load(handle)
    keys = {}
    for entry in document.get("keys", []):
        if entry.get("algorithm") == "Ed25519":
            keys[entry["keyId"]] = base64.b64decode(entry["publicKey"], validate=True)
    return keys


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    commands = parser.add_subparsers(dest="command", required=True)

    check = commands.add_parser("verify", help="check a manifest's signature")
    check.add_argument("--manifest", required=True)
    check.add_argument("--keys", required=True)
    check.add_argument("--json", action="store_true", help="report as one JSON object")

    show = commands.add_parser("canonicalize", help="print a manifest's canonical form")
    show.add_argument("--manifest", required=True)

    arguments = parser.parse_args(argv)

    with open(arguments.manifest, "rb") as handle:
        data = handle.read()

    if arguments.command == "canonicalize":
        try:
            sys.stdout.buffer.write(canonical_bytes(parse(data)))
        except Refused as refusal:
            print(f"refused: {refusal}", file=sys.stderr)
            return 2
        return 0

    report = {"valid": False, "keyId": None, "problem": None, "canonical": False, "artifacts": 0}
    try:
        key_id, manifest = verify(data, load_keys(arguments.keys))
        report.update(
            valid=True,
            keyId=key_id,
            canonical=canonical_bytes(manifest) == data,
            artifacts=len(manifest.get("artifacts", [])),
        )
    except Refused as refusal:
        report["problem"] = str(refusal)

    if arguments.json:
        print(json.dumps(report))
    else:
        print("valid" if report["valid"] else f"refused: {report['problem']}")
    return 0 if report["valid"] else 1


if __name__ == "__main__":
    sys.exit(main())