http-req.json

{
  "received_at_unix": 1781486729,
  "authority": "asnokaze.com",
  "headers": {
    "signature": "sig=:+wkpd7nQVTXX2qrvPUiDm2ZXU5Uzdgi759hZgzZmgViGue+zgxa12+ysInY1IBEIjqFUpgG8YzJ8FekBZ6JGAw==:",
    "signature-input": "sig=(\"@authority\" \"signature-agent\");created=1781486729;keyid=\"e3vpiy0B6M1Wdxnizw3dqRSgpqS6SXM2qiQ6HtUwZ5g\";alg=\"ed25519\";expires=1781490329;nonce=\"NcuWdzP41zzBqJr0SAr4RJQQZaKt72aCN8rk_Ee7z_vgaeD-4l7G0PMisYEZnRftOxmoNoFnM-fiJzNaea2ZLg\";tag=\"web-bot-auth\"",
    "signature-agent": "\"https://ahrefs.com\""
  }
}

verify.py

#!/usr/bin/env python3
import argparse
import base64
import hashlib
import json
import sys
import urllib.error
import urllib.parse
import urllib.request

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


DIRECTORY_MEDIA_TYPE = "application/http-message-signatures-directory+json"


class VerificationError(Exception):
    pass


def log_step(number, message):
    print(f"[step {number}] {message}")


def log_ok(message):
    print(f"  [ok] {message}")


def log_info(message):
    print(f"  [info] {message}")


def log_fail(message):
    print(f"  [fail] {message}")


def b64url_decode(value):
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def b64url_encode(value):
    return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=")


def split_top_level(value, separator):
    parts = []
    start = 0
    depth = 0
    in_string = False
    escaped = False

    for index, char in enumerate(value):
        if in_string:
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == '"':
                in_string = False
            continue

        if char == '"':
            in_string = True
        elif char == "(":
            depth += 1
        elif char == ")":
            depth -= 1
        elif char == separator and depth == 0:
            part = value[start:index].strip()
            if part:
                parts.append(part)
            start = index + 1

    part = value[start:].strip()
    if part:
        parts.append(part)
    return parts


def parse_dictionary(value):
    result = {}
    for member in split_top_level(value, ","):
        if "=" not in member:
            raise VerificationError(f"dictionary member has no '=': {member}")
        label, raw_value = member.split("=", 1)
        label = label.strip()
        if not label:
            raise VerificationError("dictionary member label is empty")
        result[label] = raw_value.strip()
    return result


def parse_sf_string(value):
    value = value.strip()
    if len(value) < 2 or value[0] != '"' or value[-1] != '"':
        raise VerificationError(f"not an sf-string: {value}")

    output = []
    escaped = False
    for char in value[1:-1]:
        if escaped:
            output.append(char)
            escaped = False
        elif char == "\\":
            escaped = True
        else:
            output.append(char)

    if escaped:
        raise VerificationError(f"unterminated escape in sf-string: {value}")
    return "".join(output)


def find_matching_paren(value, start):
    if value[start] != "(":
        raise VerificationError("inner list does not start with '('")

    depth = 0
    in_string = False
    escaped = False
    for index in range(start, len(value)):
        char = value[index]
        if in_string:
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == '"':
                in_string = False
            continue

        if char == '"':
            in_string = True
        elif char == "(":
            depth += 1
        elif char == ")":
            depth -= 1
            if depth == 0:
                return index

    raise VerificationError("inner list is missing closing ')'")


def parse_parameter_value(raw_value):
    raw_value = raw_value.strip()
    if raw_value.startswith('"'):
        return parse_sf_string(raw_value)
    if raw_value.startswith(":") and raw_value.endswith(":"):
        return raw_value[1:-1]
    if raw_value == "?1":
        return True
    if raw_value == "?0":
        return False
    if raw_value.lstrip("-").isdigit():
        return int(raw_value)
    return raw_value


def parse_parameters(raw_parameters):
    parameters = {}
    if not raw_parameters:
        return parameters

    for part in split_top_level(raw_parameters, ";"):
        if "=" in part:
            name, raw_value = part.split("=", 1)
            parameters[name.strip()] = parse_parameter_value(raw_value)
        else:
            parameters[part.strip()] = True
    return parameters


def parse_component(component_raw):
    parts = split_top_level(component_raw, ";")
    if not parts:
        raise VerificationError("empty component identifier")
    name = parse_sf_string(parts[0]).lower()
    params_raw = ""
    if len(parts) > 1:
        params_raw = ";" + ";".join(parts[1:])
    return {
        "raw": component_raw,
        "name": name,
        "params": parse_parameters(params_raw),
    }


def parse_signature_input_value(value):
    value = value.strip()
    close_index = find_matching_paren(value, 0)
    component_list_raw = value[: close_index + 1]
    raw_parameters = value[close_index + 1 :]
    component_content = component_list_raw[1:-1].strip()
    component_items = split_top_level(component_content, " ") if component_content else []
    components = [parse_component(item) for item in component_items]
    return {
        "raw": value,
        "components": components,
        "parameters": parse_parameters(raw_parameters),
    }


def parse_signature_header(value):
    signatures = {}
    for label, raw_value in parse_dictionary(value).items():
        if not (raw_value.startswith(":") and raw_value.endswith(":")):
            raise VerificationError(f"signature value for {label} is not a byte sequence")
        signatures[label] = raw_value[1:-1]
    return signatures


def parse_signature_input_header(value):
    return {
        label: parse_signature_input_value(raw_value)
        for label, raw_value in parse_dictionary(value).items()
    }


def parse_signature_agent(value):
    value = value.strip()
    if value.startswith('"'):
        origin = parse_sf_string(value)
    else:
        members = parse_dictionary(value)
        if "sig" in members:
            origin = parse_sf_string(members["sig"])
        else:
            first_value = next(iter(members.values()))
            origin = parse_sf_string(first_value)

    parsed = urllib.parse.urlparse(origin)
    if parsed.scheme != "https" or not parsed.netloc:
        raise VerificationError(f"signature-agent origin must be an https origin: {origin}")
    if parsed.path not in ("", "/") or parsed.params or parsed.query or parsed.fragment:
        raise VerificationError(f"signature-agent must be an origin, not a URL with path/query: {origin}")

    return origin, f"{parsed.scheme}://{parsed.netloc}/.well-known/http-message-signatures-directory"


def fetch_directory(directory_url, timeout):
    request = urllib.request.Request(
        directory_url,
        headers={
            "Accept": DIRECTORY_MEDIA_TYPE,
            "User-Agent": "web-bot-auth-verifier/1.0",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            body = response.read()
            headers = {key.lower(): value for key, value in response.headers.items()}
            status = response.status
    except urllib.error.URLError as exc:
        raise VerificationError(f"failed to fetch signature directory: {exc}") from exc

    try:
        payload = json.loads(body.decode("utf-8"))
    except json.JSONDecodeError as exc:
        raise VerificationError(f"signature directory is not valid JSON: {exc}") from exc

    return status, headers, payload


def jwk_thumbprint(jwk):
    required = {"crv": jwk.get("crv"), "kty": jwk.get("kty"), "x": jwk.get("x")}
    canonical = json.dumps(required, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return b64url_encode(hashlib.sha256(canonical).digest())


def find_key(directory_payload, keyid):
    for key in directory_payload.get("keys", []):
        if key.get("kid") == keyid:
            return key
    raise VerificationError(f"keyid not found in signature directory: {keyid}")


def ed25519_public_key_from_jwk(jwk):
    if jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519":
        raise VerificationError("JWK is not an Ed25519 OKP key")
    if jwk.get("use") not in (None, "sig"):
        raise VerificationError(f"JWK use is not 'sig': {jwk.get('use')}")
    return Ed25519PublicKey.from_public_bytes(b64url_decode(jwk["x"]))


def verify_ed25519(public_key, signature_base64, signature_base):
    try:
        signature = base64.b64decode(signature_base64, validate=True)
    except ValueError as exc:
        raise VerificationError(f"signature is not valid base64: {exc}") from exc

    try:
        public_key.verify(signature, signature_base.encode("ascii"))
    except UnicodeEncodeError as exc:
        raise VerificationError("signature base contains non-ASCII data") from exc
    except InvalidSignature as exc:
        raise VerificationError("signature cryptographic verification failed") from exc


def value_for_request_component(component, request):
    name = component["name"]
    if component["params"]:
        raise VerificationError(f"unsupported request component parameters: {component['raw']}")

    if name == "@authority":
        authority = request.get("authority")
        if not authority:
            raise VerificationError("request authority is missing")
        return authority

    headers = {key.lower(): value for key, value in request.get("headers", {}).items()}
    if name not in headers:
        raise VerificationError(f"covered request header is missing: {name}")
    return headers[name]


def value_for_directory_component(component, directory_request_authority, directory_response_headers):
    name = component["name"]
    params = component["params"]

    if name == "@authority":
        return directory_request_authority

    if params.get("req"):
        raise VerificationError(f"cannot reconstruct covered directory request header: {component['raw']}")

    if name not in directory_response_headers:
        raise VerificationError(f"covered directory response header is missing: {name}")
    return directory_response_headers[name]


def build_signature_base(signature_input, component_value_fn):
    lines = []
    for component in signature_input["components"]:
        lines.append(f"{component['raw']}: {component_value_fn(component)}")
    lines.append(f'"@signature-params": {signature_input["raw"]}')
    return "\n".join(lines)


def print_signature_base(signature_base):
    for line in signature_base.splitlines():
        print(f"    {line}")


def verify_directory_binding(directory_url, response_headers, payload):
    parsed_url = urllib.parse.urlparse(directory_url)
    request_authority = parsed_url.netloc
    signature_header = response_headers.get("signature")
    signature_input_header = response_headers.get("signature-input")

    if not signature_header or not signature_input_header:
        raise VerificationError("signature directory response is missing Signature or Signature-Input")

    signatures = parse_signature_header(signature_header)
    signature_inputs = parse_signature_input_header(signature_input_header)
    valid_labels = []

    for label, signature_input in signature_inputs.items():
        params = signature_input["parameters"]
        if params.get("tag") != "http-message-signatures-directory":
            continue
        if params.get("alg") != "ed25519":
            continue
        if label not in signatures:
            raise VerificationError(f"directory Signature is missing label: {label}")

        key = find_key(payload, params.get("keyid"))
        public_key = ed25519_public_key_from_jwk(key)
        signature_base = build_signature_base(
            signature_input,
            lambda component: value_for_directory_component(
                component, request_authority, response_headers
            ),
        )
        verify_ed25519(public_key, signatures[label], signature_base)
        valid_labels.append(label)

    if not valid_labels:
        raise VerificationError("no valid http-message-signatures-directory binding was found")

    return valid_labels


def verify_request(request, key):
    headers = {key.lower(): value for key, value in request.get("headers", {}).items()}
    if "signature" not in headers:
        raise VerificationError("request is missing Signature")
    if "signature-input" not in headers:
        raise VerificationError("request is missing Signature-Input")
    if "signature-agent" not in headers:
        raise VerificationError("request is missing Signature-Agent")

    signatures = parse_signature_header(headers["signature"])
    signature_inputs = parse_signature_input_header(headers["signature-input"])
    if len(signature_inputs) != 1:
        raise VerificationError("expected exactly one request signature input")

    label, signature_input = next(iter(signature_inputs.items()))
    if label not in signatures:
        raise VerificationError(f"Signature is missing label from Signature-Input: {label}")

    params = signature_input["parameters"]
    if params.get("tag") != "web-bot-auth":
        raise VerificationError(f"unexpected signature tag: {params.get('tag')}")
    if params.get("alg") != "ed25519":
        raise VerificationError(f"unsupported signature alg: {params.get('alg')}")
    if params.get("keyid") != key.get("kid"):
        raise VerificationError("request keyid does not match selected directory key")

    received_at = request.get("received_at_unix")
    if received_at is None:
        raise VerificationError("received_at_unix is missing")
    if "created" in params and received_at < params["created"]:
        raise VerificationError("request was received before signature created time")
    if "expires" in params and received_at > params["expires"]:
        raise VerificationError("request was received after signature expiry time")

    signature_base = build_signature_base(
        signature_input,
        lambda component: value_for_request_component(component, request),
    )
    public_key = ed25519_public_key_from_jwk(key)
    verify_ed25519(public_key, signatures[label], signature_base)

    return label, signature_input, signature_base


def verify(input_path, timeout):
    log_step(1, f"load input JSON: {input_path}")
    with open(input_path, "r", encoding="utf-8") as file:
        request = json.load(file)
    log_ok("JSON loaded")
    log_info(f"authority = {request.get('authority')}")
    log_info(f"received_at_unix = {request.get('received_at_unix')}")

    headers = {key.lower(): value for key, value in request.get("headers", {}).items()}

    log_step(2, "parse Signature-Agent and locate signature directory")
    origin, directory_url = parse_signature_agent(headers.get("signature-agent", ""))
    log_ok(f"signature-agent origin = {origin}")
    log_ok(f"directory_url = {directory_url}")

    log_step(3, "parse request Signature and Signature-Input")
    request_signatures = parse_signature_header(headers.get("signature", ""))
    request_signature_inputs = parse_signature_input_header(headers.get("signature-input", ""))
    if len(request_signature_inputs) != 1:
        raise VerificationError("expected exactly one request signature input")
    request_label, request_signature_input = next(iter(request_signature_inputs.items()))
    request_params = request_signature_input["parameters"]
    log_ok(f"label = {request_label}")
    log_ok("covered components = " + ", ".join(c["raw"] for c in request_signature_input["components"]))
    log_ok(f"keyid = {request_params.get('keyid')}")
    log_ok(f"alg = {request_params.get('alg')}")
    log_ok(f"tag = {request_params.get('tag')}")
    if request_label not in request_signatures:
        raise VerificationError(f"Signature is missing label from Signature-Input: {request_label}")

    log_step(4, "fetch HTTP Message Signatures Directory")
    status, directory_headers, directory_payload = fetch_directory(directory_url, timeout)
    log_ok(f"HTTP status = {status}")
    content_type = directory_headers.get("content-type", "")
    log_ok(f"content-type = {content_type}")
    if not content_type.lower().startswith(DIRECTORY_MEDIA_TYPE):
        raise VerificationError(f"unexpected directory content-type: {content_type}")
    keys = directory_payload.get("keys", [])
    log_ok(f"keys = {len(keys)}")

    log_step(5, "verify directory response binding signature")
    valid_bindings = verify_directory_binding(directory_url, directory_headers, directory_payload)
    log_ok("valid directory binding labels = " + ", ".join(valid_bindings))

    log_step(6, "select request verification key")
    keyid = request_params.get("keyid")
    key = find_key(directory_payload, keyid)
    log_ok("key found in directory")
    expected_thumbprint = jwk_thumbprint(key)
    if expected_thumbprint != key.get("kid"):
        raise VerificationError("JWK kid does not match its SHA-256 thumbprint")
    log_ok("JWK kid matches SHA-256 JWK thumbprint")
    log_ok(f"kty/crv/use = {key.get('kty')}/{key.get('crv')}/{key.get('use')}")

    log_step(7, "verify request time window")
    received_at = request.get("received_at_unix")
    created = request_params.get("created")
    expires = request_params.get("expires")
    log_info(f"received_at = {received_at}")
    log_info(f"created = {created}")
    log_info(f"expires = {expires}")
    if created is not None and received_at < created:
        raise VerificationError("request was received before signature created time")
    if expires is not None and received_at > expires:
        raise VerificationError("request was received after signature expiry time")
    log_ok("received time is within the signature time window")

    log_step(8, "build request signature base")
    _, _, signature_base = verify_request(request, key)
    print_signature_base(signature_base)
    log_ok("signature base built")

    log_step(9, "verify request Ed25519 signature")
    log_ok("request signature is VALID")


def main():
    parser = argparse.ArgumentParser(
        description="Verify a Web-Bot-Auth request JSON with step-by-step logs."
    )
    parser.add_argument(
        "input",
        nargs="?",
        default="http-req.json",
        help="request JSON file (default: http-req.json)",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=20.0,
        help="HTTP timeout for signature directory fetch in seconds",
    )
    args = parser.parse_args()

    try:
        verify(args.input, args.timeout)
    except (OSError, json.JSONDecodeError, VerificationError) as exc:
        log_fail(str(exc))
        return 1
    return 0


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