crank/scripts/validate-capability-run.py

#!/usr/bin/env python3
"""Validate a sanitized evidence candidate against capability-baseline $defs/run."""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any


MAX_INPUT_BYTES = 65_536


class ValidationError(Exception):
    def __init__(self, pointer: str, message: str) -> None:
        super().__init__(message)
        self.pointer = pointer
        self.message = message


def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValidationError('/', 'duplicate JSON key')
        result[key] = value
    return result


def load_json(path: Path) -> Any:
    try:
        raw = path.read_bytes()
    except OSError as error:
        raise ValidationError('/', 'cannot read input') from error
    if len(raw) > MAX_INPUT_BYTES:
        raise ValidationError('/', 'input exceeds 64 KiB')
    try:
        return json.loads(raw.decode('utf-8'), object_pairs_hook=reject_duplicates)
    except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as error:
        raise ValidationError('/', 'invalid JSON') from error


def pointer_child(pointer: str, key: str | int) -> str:
    escaped = str(key).replace('~', '~0').replace('/', '~1')
    return f'{pointer}/{escaped}'


def type_matches(value: Any, expected: str) -> bool:
    if expected == 'object':
        return isinstance(value, dict)
    if expected == 'array':
        return isinstance(value, list)
    if expected == 'string':
        return isinstance(value, str)
    if expected == 'boolean':
        return type(value) is bool
    return False


def validate(value: Any, schema: dict[str, Any], pointer: str = '') -> None:
    current = pointer or '/'
    expected_type = schema.get('type')
    if expected_type and not type_matches(value, expected_type):
        raise ValidationError(current, f'expected {expected_type}')

    if 'const' in schema and value != schema['const']:
        raise ValidationError(current, 'does not match const')
    if 'enum' in schema and value not in schema['enum']:
        raise ValidationError(current, 'does not match enum')

    if isinstance(value, str):
        if len(value) < schema.get('minLength', 0):
            raise ValidationError(current, 'string is too short')
        if len(value) > schema.get('maxLength', sys.maxsize):
            raise ValidationError(current, 'string is too long')
        pattern = schema.get('pattern')
        if pattern and not re.fullmatch(pattern, value):
            raise ValidationError(current, 'string does not match pattern')

    if isinstance(value, list):
        if len(value) < schema.get('minItems', 0):
            raise ValidationError(current, 'array has too few items')
        if len(value) > schema.get('maxItems', sys.maxsize):
            raise ValidationError(current, 'array has too many items')
        if schema.get('uniqueItems') and len({json.dumps(item, sort_keys=True) for item in value}) != len(value):
            raise ValidationError(current, 'array items are not unique')
        item_schema = schema.get('items')
        if item_schema:
            for index, item in enumerate(value):
                validate(item, item_schema, pointer_child(pointer, index))

    if isinstance(value, dict):
        required = schema.get('required', [])
        for key in required:
            if key not in value:
                raise ValidationError(current, f'missing required property {key}')
        properties = schema.get('properties', {})
        if schema.get('additionalProperties') is False:
            unexpected = set(value) - set(properties)
            if unexpected:
                raise ValidationError(current, f'unexpected property {sorted(unexpected)[0]}')
        if len(value) > schema.get('maxProperties', sys.maxsize):
            raise ValidationError(current, 'object has too many properties')
        for key, item in value.items():
            item_schema = properties.get(key)
            if item_schema:
                validate(item, item_schema, pointer_child(pointer, key))


def run_schema(schema: Any) -> dict[str, Any]:
    if not isinstance(schema, dict):
        raise ValidationError('/schema', 'schema root must be an object')
    definitions = schema.get('$defs')
    if not isinstance(definitions, dict):
        raise ValidationError('/schema/$defs', 'definitions are missing')
    definition = definitions.get('run')
    if not isinstance(definition, dict):
        raise ValidationError('/schema/$defs/run', 'run definition is missing')
    return definition


def validate_accepted_semantics(candidate: Any) -> None:
    """Fail closed: accepted evidence must be an automated passing run."""
    if not isinstance(candidate, dict):
        raise ValidationError('/', 'accepted evidence candidate must be an object')
    if candidate.get('accepted') is not True:
        raise ValidationError('/accepted', 'accepted evidence is required')
    if candidate.get('execution_verdict') != 'pass':
        raise ValidationError('/execution_verdict', 'accepted evidence requires execution_verdict=pass')
    if candidate.get('evidence_mode') != 'automated':
        raise ValidationError('/evidence_mode', 'accepted evidence requires evidence_mode=automated')


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--schema', required=True)
    parser.add_argument('--candidate', required=True)
    parser.add_argument('--require-accepted', action='store_true')
    args = parser.parse_args()

    try:
        schema = load_json(Path(args.schema))
        candidate = load_json(Path(args.candidate))
        validate(candidate, run_schema(schema))
        if args.require_accepted:
            validate_accepted_semantics(candidate)
    except ValidationError as error:
        print(f'INVALID_CAPABILITY_RUN pointer={error.pointer} reason={error.message}', file=sys.stderr)
        return 1

    print('Capability evidence candidate matches $defs/run.')
    return 0


if __name__ == '__main__':
    raise SystemExit(main())