36 lines
1.5 KiB
Python
36 lines
1.5 KiB
Python
import json
|
|
from importlib.resources import files
|
|
|
|
from jsonschema import Draft202012Validator
|
|
from jsonschema.exceptions import ValidationError
|
|
|
|
from viano_bug_analysis.package import PackageReadError
|
|
from viano_bug_analysis.package.manifest import FORMAT_CHECKER, parse_manifest, safe_path
|
|
|
|
from .errors import CaseBuildError
|
|
|
|
|
|
def validate(document: dict, contract: str, path: str | None = None) -> None:
|
|
schema = json.loads(files("viano_bug_analysis.schemas").joinpath(f"{contract}-v1.schema.json").read_text())
|
|
try:
|
|
Draft202012Validator(schema, format_checker=FORMAT_CHECKER).validate(document)
|
|
except ValidationError as error:
|
|
raise CaseBuildError("INPUT_SCHEMA" if contract != "case" else "CASE_SCHEMA", error.message, path) from error
|
|
|
|
|
|
def read_json(reader, path: str, contract: str, limit: int) -> dict:
|
|
require_file(reader, path)
|
|
try:
|
|
with reader._archive.open(path) as stream:
|
|
data = reader.validator._read_bounded(stream, min(limit, reader.validator.limits.max_file_bytes))
|
|
document = parse_manifest(data)
|
|
validate(document, contract, path)
|
|
return document
|
|
except (ValueError, RecursionError, PackageReadError) as error:
|
|
raise CaseBuildError("INPUT_JSON", str(error), path) from error
|
|
|
|
|
|
def require_file(reader, path: str) -> None:
|
|
if not safe_path(path) or path == "manifest.json" or path not in reader.files:
|
|
raise CaseBuildError("INPUT_REFERENCE", "Expected a safe existing payload file", path)
|