feat(perception): add autonomous vegetation shadow lab
This commit is contained in:
@@ -326,6 +326,7 @@ def canonical_laboratory_adapters() -> dict[str, LaboratoryAdapter]:
|
||||
"experimental.e47-semantic-slam-shadow/v1": _run_e47,
|
||||
"experimental.m48s-fixed-class-detector/v1": _run_m48s_fixed_class_detector,
|
||||
"experimental.m48t-risk-quality-temporal/v1": _run_m48t_risk_quality_temporal,
|
||||
"experimental.lab-v1-vegetation-shadow/v1": _run_lab_v1_vegetation_shadow,
|
||||
}
|
||||
|
||||
|
||||
@@ -382,6 +383,28 @@ def _run_m48t_risk_quality_temporal(
|
||||
)
|
||||
|
||||
|
||||
def _run_lab_v1_vegetation_shadow(
|
||||
request: LaboratoryRunRequest,
|
||||
) -> LaboratoryAdapterResult:
|
||||
from k1link.laboratory.vegetation_shadow_lab import seal_vegetation_shadow_lab
|
||||
|
||||
result_root = seal_vegetation_shadow_lab(
|
||||
ddrnet_goose_root=request.inputs["ddrnet_goose_root"],
|
||||
ppliteseg_goose_root=request.inputs["ppliteseg_goose_root"],
|
||||
ddrnet_ravnoves_root=request.inputs["ddrnet_ravnoves_root"],
|
||||
ppliteseg_ravnoves_root=request.inputs["ppliteseg_ravnoves_root"],
|
||||
output_root=request.output_root,
|
||||
)
|
||||
manifest = _object(
|
||||
json.loads((result_root / "result.json").read_text(encoding="utf-8")),
|
||||
"vegetation shadow result",
|
||||
)
|
||||
result_id = manifest.get("result_id")
|
||||
if not isinstance(result_id, str):
|
||||
raise LaboratoryExecutionError("vegetation shadow result_id is invalid")
|
||||
return LaboratoryAdapterResult(result_root=result_root, result_id=result_id)
|
||||
|
||||
|
||||
def _run_m48_small_static_passage_regression(
|
||||
request: LaboratoryRunRequest,
|
||||
) -> LaboratoryAdapterResult:
|
||||
|
||||
@@ -0,0 +1,381 @@
|
||||
"""Seal GOOSE qualification and RAVNOVES vegetation shadow evidence for LAB."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import shutil
|
||||
import tempfile
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any, Final
|
||||
|
||||
LAB_SCHEMA: Final = "missioncore.lab-v1-vegetation-shadow/v1"
|
||||
WORKER_SCHEMA: Final = "missioncore.lab-v1-goose-vegetation-run/v1"
|
||||
RESULT_PREFIX: Final = "lab-v1-vegetation-shadow-"
|
||||
_CANDIDATES: Final = ("ddrnet", "ppliteseg")
|
||||
_MODES: Final = ("goose", "ravnoves")
|
||||
_IMAGE_KEYS: Final = (
|
||||
"source",
|
||||
"prediction_semantic",
|
||||
"policy_urban",
|
||||
"policy_rural",
|
||||
"policy_offroad",
|
||||
"truth_semantic",
|
||||
)
|
||||
|
||||
|
||||
class VegetationShadowLabError(ValueError):
|
||||
"""Raised when Worker evidence cannot be sealed without changing its meaning."""
|
||||
|
||||
|
||||
def canonical_json(value: object) -> bytes:
|
||||
return json.dumps(
|
||||
value,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
allow_nan=False,
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def sha256_path(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _object(value: object, label: str) -> dict[str, Any]:
|
||||
if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
|
||||
raise VegetationShadowLabError(f"{label} must be an object")
|
||||
return value
|
||||
|
||||
|
||||
def _read_worker_result(root: Path, *, candidate: str, mode: str) -> dict[str, Any]:
|
||||
if root.is_symlink() or not root.is_dir():
|
||||
raise VegetationShadowLabError(f"{candidate}/{mode} result root is unavailable")
|
||||
path = root / "result.json"
|
||||
if path.is_symlink() or not path.is_file() or path.stat().st_size > 1024 * 1024:
|
||||
raise VegetationShadowLabError(f"{candidate}/{mode} result document is unavailable")
|
||||
try:
|
||||
result = _object(json.loads(path.read_text("utf-8")), f"{candidate}/{mode} result")
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
raise VegetationShadowLabError(f"{candidate}/{mode} result is invalid") from exc
|
||||
candidate_value = _object(result.get("candidate"), f"{candidate}/{mode} candidate")
|
||||
authority = _object(result.get("authority"), f"{candidate}/{mode} authority")
|
||||
if (
|
||||
result.get("schema_version") != WORKER_SCHEMA
|
||||
or result.get("mode") != mode
|
||||
or candidate_value.get("candidate_key") != candidate
|
||||
or authority.get("navigation_accepted") is not False
|
||||
or authority.get("safety_accepted") is not False
|
||||
or authority.get("actuation_accepted") is not False
|
||||
or authority.get("camera_semantics_can_clear_rigid_geometry") is not False
|
||||
):
|
||||
raise VegetationShadowLabError(f"{candidate}/{mode} contract changed")
|
||||
return result
|
||||
|
||||
|
||||
def _case_map(result: dict[str, Any], label: str) -> dict[str, dict[str, Any]]:
|
||||
values = result.get("visual_cases")
|
||||
if not isinstance(values, list) or len(values) != 12:
|
||||
raise VegetationShadowLabError(f"{label} must contain exactly 12 visual cases")
|
||||
rows: dict[str, dict[str, Any]] = {}
|
||||
for raw in values:
|
||||
row = _object(raw, f"{label} visual case")
|
||||
case_id = row.get("case_id")
|
||||
if not isinstance(case_id, str) or not case_id or case_id in rows:
|
||||
raise VegetationShadowLabError(f"{label} case identity changed")
|
||||
rows[case_id] = row
|
||||
return rows
|
||||
|
||||
|
||||
def _file_from_case(
|
||||
root: Path,
|
||||
case: dict[str, Any],
|
||||
key: str,
|
||||
*,
|
||||
required: bool = True,
|
||||
) -> tuple[Path, str] | None:
|
||||
files = _object(case.get("files"), "visual case files")
|
||||
raw = files.get(key)
|
||||
if raw is None and not required:
|
||||
return None
|
||||
descriptor = _object(raw, f"visual case {key}")
|
||||
relative = descriptor.get("relative_path")
|
||||
expected_sha256 = descriptor.get("sha256")
|
||||
if not isinstance(relative, str) or not isinstance(expected_sha256, str):
|
||||
raise VegetationShadowLabError(f"visual case {key} proof is invalid")
|
||||
posix = PurePosixPath(relative)
|
||||
if posix.is_absolute() or str(posix) != relative or any(
|
||||
part in {"", ".", ".."} for part in posix.parts
|
||||
):
|
||||
raise VegetationShadowLabError(f"visual case {key} path is invalid")
|
||||
path = root.joinpath(*posix.parts)
|
||||
if path.is_symlink() or not path.is_file() or not path.resolve().is_relative_to(root.resolve()):
|
||||
raise VegetationShadowLabError(f"visual case {key} file is unavailable")
|
||||
if sha256_path(path) != expected_sha256:
|
||||
raise VegetationShadowLabError(f"visual case {key} digest changed")
|
||||
return path, expected_sha256
|
||||
|
||||
|
||||
def _copy_artifact(
|
||||
source: Path,
|
||||
destination_root: Path,
|
||||
relative: str,
|
||||
artifacts: list[dict[str, object]],
|
||||
*,
|
||||
role: str,
|
||||
media_type: str,
|
||||
) -> dict[str, object]:
|
||||
destination = destination_root.joinpath(*PurePosixPath(relative).parts)
|
||||
destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
shutil.copyfile(source, destination)
|
||||
digest = sha256_path(destination)
|
||||
descriptor = {
|
||||
"role": role,
|
||||
"path": relative,
|
||||
"byte_length": destination.stat().st_size,
|
||||
"sha256": digest,
|
||||
"media_type": media_type,
|
||||
}
|
||||
artifacts.append(descriptor)
|
||||
return descriptor
|
||||
|
||||
|
||||
def _selected_candidate(results: dict[tuple[str, str], dict[str, Any]]) -> str:
|
||||
def rank(candidate: str) -> tuple[float, float]:
|
||||
validation = results[(candidate, "goose")]
|
||||
metrics = _object(validation.get("metrics"), f"{candidate} metrics")
|
||||
timing = _object(validation.get("timing"), f"{candidate} timing")
|
||||
vegetation_iou = metrics.get("vegetation_mean_iou")
|
||||
p95_ms = timing.get("latency_ms_p95")
|
||||
if not isinstance(vegetation_iou, (int, float)) or not isinstance(p95_ms, (int, float)):
|
||||
raise VegetationShadowLabError(f"{candidate} qualification metrics are incomplete")
|
||||
return float(vegetation_iou), -float(p95_ms)
|
||||
|
||||
return max(_CANDIDATES, key=rank)
|
||||
|
||||
|
||||
def _validation_metric_summary(result: dict[str, Any], candidate: str) -> dict[str, object]:
|
||||
metrics = _object(result.get("metrics"), f"{candidate} metrics")
|
||||
return {
|
||||
"mean_iou_percent": metrics.get("mean_iou_percent"),
|
||||
"published_mean_iou_percent": metrics.get("published_mean_iou_percent"),
|
||||
"vegetation_mean_iou": metrics.get("vegetation_mean_iou"),
|
||||
}
|
||||
|
||||
|
||||
def seal_vegetation_shadow_lab(
|
||||
*,
|
||||
ddrnet_goose_root: Path,
|
||||
ppliteseg_goose_root: Path,
|
||||
ddrnet_ravnoves_root: Path,
|
||||
ppliteseg_ravnoves_root: Path,
|
||||
output_root: Path,
|
||||
) -> Path:
|
||||
roots = {
|
||||
("ddrnet", "goose"): ddrnet_goose_root.resolve(),
|
||||
("ppliteseg", "goose"): ppliteseg_goose_root.resolve(),
|
||||
("ddrnet", "ravnoves"): ddrnet_ravnoves_root.resolve(),
|
||||
("ppliteseg", "ravnoves"): ppliteseg_ravnoves_root.resolve(),
|
||||
}
|
||||
results = {
|
||||
key: _read_worker_result(root, candidate=key[0], mode=key[1])
|
||||
for key, root in roots.items()
|
||||
}
|
||||
cases = {key: _case_map(result, f"{key[0]}/{key[1]}") for key, result in results.items()}
|
||||
for mode in _MODES:
|
||||
if cases[("ddrnet", mode)].keys() != cases[("ppliteseg", mode)].keys():
|
||||
raise VegetationShadowLabError(f"{mode} candidate case islands differ")
|
||||
selected = _selected_candidate(results)
|
||||
|
||||
output_root.mkdir(mode=0o700, parents=True, exist_ok=True)
|
||||
temporary = Path(tempfile.mkdtemp(prefix=".lab-v1-vegetation-", dir=output_root))
|
||||
artifacts: list[dict[str, object]] = []
|
||||
catalogs: dict[str, list[dict[str, object]]] = {"goose": [], "ravnoves": []}
|
||||
try:
|
||||
for mode in _MODES:
|
||||
for case_id in sorted(cases[("ddrnet", mode)]):
|
||||
ddr_case = cases[("ddrnet", mode)][case_id]
|
||||
pplite_case = cases[("ppliteseg", mode)][case_id]
|
||||
row: dict[str, object] = {
|
||||
"case_id": case_id,
|
||||
"source_kind": mode,
|
||||
"width": ddr_case.get("source_width"),
|
||||
"height": ddr_case.get("source_height"),
|
||||
"center_crop_xyxy": ddr_case.get("center_crop_xyxy"),
|
||||
"outside_crop_state": ddr_case.get("outside_crop_state"),
|
||||
"assets": {},
|
||||
}
|
||||
asset_map = _object(row["assets"], "sealed assets")
|
||||
sources: list[tuple[str, str, dict[str, Any], str]] = [
|
||||
("source", "ddrnet", ddr_case, "source"),
|
||||
("ddrnet", "ddrnet", ddr_case, "prediction_semantic"),
|
||||
("ppliteseg", "ppliteseg", pplite_case, "prediction_semantic"),
|
||||
]
|
||||
if mode == "goose":
|
||||
sources.append(("truth", "ddrnet", ddr_case, "truth_semantic"))
|
||||
else:
|
||||
selected_case = cases[(selected, mode)][case_id]
|
||||
sources.extend(
|
||||
(preset, selected, selected_case, f"policy_{preset}")
|
||||
for preset in ("urban", "rural", "offroad")
|
||||
)
|
||||
for asset_key, candidate, case, worker_key in sources:
|
||||
resolved = _file_from_case(roots[(candidate, mode)], case, worker_key)
|
||||
assert resolved is not None
|
||||
source_path, _ = resolved
|
||||
relative = f"visual/{mode}/{case_id}/{asset_key}.png"
|
||||
descriptor = _copy_artifact(
|
||||
source_path,
|
||||
temporary,
|
||||
relative,
|
||||
artifacts,
|
||||
role=f"visual-{mode}-{asset_key}",
|
||||
media_type="image/png",
|
||||
)
|
||||
asset_map[asset_key] = {
|
||||
"path": descriptor["path"],
|
||||
"sha256": descriptor["sha256"],
|
||||
}
|
||||
catalogs[mode].append(row)
|
||||
|
||||
worker_proofs: dict[str, dict[str, object]] = {}
|
||||
for candidate in _CANDIDATES:
|
||||
for mode in _MODES:
|
||||
source = roots[(candidate, mode)] / "result.json"
|
||||
relative = f"worker/{candidate}-{mode}.json"
|
||||
descriptor = _copy_artifact(
|
||||
source,
|
||||
temporary,
|
||||
relative,
|
||||
artifacts,
|
||||
role="worker-result",
|
||||
media_type="application/json",
|
||||
)
|
||||
worker_proofs[f"{candidate}_{mode}"] = {
|
||||
"result_id": results[(candidate, mode)].get("result_id"),
|
||||
"path": relative,
|
||||
"sha256": descriptor["sha256"],
|
||||
}
|
||||
|
||||
candidate_metrics: dict[str, object] = {}
|
||||
for candidate in _CANDIDATES:
|
||||
validation = results[(candidate, "goose")]
|
||||
shadow = results[(candidate, "ravnoves")]
|
||||
candidate_metrics[candidate] = {
|
||||
"loaded_model_name": _object(validation["candidate"], "candidate").get(
|
||||
"loaded_model_name"
|
||||
),
|
||||
"checkpoint_sha256": _object(validation["candidate"], "candidate").get(
|
||||
"checkpoint_sha256"
|
||||
),
|
||||
# Detailed per-class rows remain immutable in worker_proofs. The
|
||||
# top-level LAB manifest carries only the UI/index summary.
|
||||
"validation_metrics": _validation_metric_summary(validation, candidate),
|
||||
"validation_timing": validation.get("timing"),
|
||||
"shadow_timing": shadow.get("timing"),
|
||||
"resource": shadow.get("resource"),
|
||||
}
|
||||
|
||||
authority = {
|
||||
"commands_enabled": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
"actuation_accepted": False,
|
||||
"camera_semantics_can_clear_rigid_geometry": False,
|
||||
}
|
||||
identity = {
|
||||
"lab_id": "lab-v1-vegetation-mission-policy",
|
||||
"source": {
|
||||
"validation_dataset": "GOOSE-2D-validation-visible-962",
|
||||
"shadow_session": "RAVNOVES00",
|
||||
"shadow_camera": "sensor.camera.right",
|
||||
"shadow_frame_count": 12,
|
||||
},
|
||||
"selected_candidate": selected,
|
||||
"candidate_metrics": candidate_metrics,
|
||||
"worker_proofs": worker_proofs,
|
||||
"visual_catalog_sha256": hashlib.sha256(canonical_json(catalogs)).hexdigest(),
|
||||
"authority": authority,
|
||||
}
|
||||
identity_sha256 = hashlib.sha256(canonical_json(identity)).hexdigest()
|
||||
result_id = f"{RESULT_PREFIX}{identity_sha256}"
|
||||
manifest = {
|
||||
"schema_version": LAB_SCHEMA,
|
||||
"result_id": result_id,
|
||||
"identity_sha256": identity_sha256,
|
||||
"created_at_utc": datetime.now(UTC).isoformat(),
|
||||
"ground_truth": False,
|
||||
"status": "visual-shadow-ready-policy-not-authorized",
|
||||
"identity": identity,
|
||||
"source": identity["source"],
|
||||
"method": {
|
||||
"completeness": "complete",
|
||||
"execution_class": "ai-inference",
|
||||
"pipeline_id": "goose-fine64-ready-weights-to-ravnoves-policy-shadow/v1",
|
||||
},
|
||||
"metrics": {"candidates": candidate_metrics},
|
||||
"decision": {
|
||||
"selected_candidate": selected,
|
||||
"visual_shadow_ready": True,
|
||||
"mission_policy_ready_for_configuration": True,
|
||||
"navigation_accepted": False,
|
||||
"production_accepted": False,
|
||||
},
|
||||
"limitations": [
|
||||
"GOOSE validation is external-domain qualification, not RAVNOVES ground truth.",
|
||||
"The RAVNOVES island is visual shadow evidence without independent labels.",
|
||||
"Vegetation semantics never clears rigid LiDAR/TGS occupancy.",
|
||||
"Undefined pixels outside the 600x600 center crop remain fail-closed.",
|
||||
],
|
||||
"authority": authority,
|
||||
"catalogs": catalogs,
|
||||
"artifacts": artifacts,
|
||||
}
|
||||
(temporary / "result.json").write_bytes(canonical_json(manifest) + b"\n")
|
||||
destination = output_root / result_id
|
||||
if destination.exists():
|
||||
raise VegetationShadowLabError("immutable vegetation LAB result already exists")
|
||||
temporary.replace(destination)
|
||||
return destination
|
||||
except Exception:
|
||||
shutil.rmtree(temporary, ignore_errors=True)
|
||||
raise
|
||||
|
||||
|
||||
def _parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--ddrnet-goose-root", type=Path, required=True)
|
||||
parser.add_argument("--ppliteseg-goose-root", type=Path, required=True)
|
||||
parser.add_argument("--ddrnet-ravnoves-root", type=Path, required=True)
|
||||
parser.add_argument("--ppliteseg-ravnoves-root", type=Path, required=True)
|
||||
parser.add_argument("--output-root", type=Path, required=True)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = _parse_args()
|
||||
destination = seal_vegetation_shadow_lab(
|
||||
ddrnet_goose_root=args.ddrnet_goose_root,
|
||||
ppliteseg_goose_root=args.ppliteseg_goose_root,
|
||||
ddrnet_ravnoves_root=args.ddrnet_ravnoves_root,
|
||||
ppliteseg_ravnoves_root=args.ppliteseg_ravnoves_root,
|
||||
output_root=args.output_root,
|
||||
)
|
||||
print(destination)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"LAB_SCHEMA",
|
||||
"RESULT_PREFIX",
|
||||
"VegetationShadowLabError",
|
||||
"seal_vegetation_shadow_lab",
|
||||
]
|
||||
@@ -138,6 +138,7 @@ from k1link.web.m49_physical_safety_playback_api import (
|
||||
)
|
||||
from k1link.web.m49_tgs_fail_closed_api import build_m49_tgs_fail_closed_router
|
||||
from k1link.web.m49_tgs_full_shadow_api import build_m49_tgs_full_shadow_router
|
||||
from k1link.web.vegetation_shadow_lab_api import build_vegetation_shadow_lab_router
|
||||
from k1link.web.map_api import (
|
||||
MapGatewayConfiguration,
|
||||
MapGatewayProxy,
|
||||
@@ -1019,6 +1020,17 @@ app.include_router(
|
||||
),
|
||||
)
|
||||
)
|
||||
app.include_router(
|
||||
build_vegetation_shadow_lab_router(
|
||||
root_provider=lambda: (
|
||||
REPOSITORY_ROOT
|
||||
/ ".runtime"
|
||||
/ "compute-experiments"
|
||||
/ "lab-v1-vegetation"
|
||||
/ "results"
|
||||
),
|
||||
)
|
||||
)
|
||||
app.include_router(
|
||||
build_m49_physical_safety_playback_router(
|
||||
root_provider=lambda: (
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
"""Read-only API for the autonomous vegetation policy shadow LAB."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import re
|
||||
from collections.abc import Callable
|
||||
from functools import lru_cache
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any, Final
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from fastapi.responses import FileResponse
|
||||
|
||||
from k1link.laboratory.evidence_registry import LaboratoryEvidenceDefinition
|
||||
from k1link.laboratory.evidence_report import (
|
||||
LaboratoryEvidenceReportError,
|
||||
verify_laboratory_evidence_result,
|
||||
)
|
||||
from k1link.laboratory.vegetation_shadow_lab import LAB_SCHEMA, RESULT_PREFIX
|
||||
|
||||
RootProvider = Callable[[], Path | None]
|
||||
RESULT_ID: Final = re.compile(rf"^{re.escape(RESULT_PREFIX)}[a-f0-9]{{64}}$")
|
||||
_MAX_DOCUMENT_BYTES: Final = 1024 * 1024
|
||||
_DEFINITION: Final = LaboratoryEvidenceDefinition(
|
||||
work_id="lab-v1-vegetation-shadow",
|
||||
runtime_relative_root=PurePosixPath("lab-v1-vegetation/results"),
|
||||
result_id_prefix="lab-v1-vegetation-shadow",
|
||||
document_name="result.json",
|
||||
result_schema_version=LAB_SCHEMA,
|
||||
)
|
||||
|
||||
|
||||
def build_vegetation_shadow_lab_router(
|
||||
*, root_provider: RootProvider = lambda: None,
|
||||
) -> APIRouter:
|
||||
router = APIRouter(
|
||||
prefix="/api/v1/laboratory/vegetation-shadow",
|
||||
tags=["laboratory"],
|
||||
)
|
||||
|
||||
@router.get("/{result_id}")
|
||||
def get_result(result_id: str) -> dict[str, object]:
|
||||
candidate = _resolve_candidate(root_provider, result_id)
|
||||
return {**copy.deepcopy(_read_verified(candidate)), "access": "read-only"}
|
||||
|
||||
@router.get("/{result_id}/assets/{asset_path:path}")
|
||||
def get_asset(result_id: str, asset_path: str) -> FileResponse:
|
||||
candidate = _resolve_candidate(root_provider, result_id)
|
||||
manifest = _read_verified(candidate)
|
||||
artifacts = manifest.get("artifacts")
|
||||
if not isinstance(artifacts, list):
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB asset not found")
|
||||
descriptor = next(
|
||||
(
|
||||
item
|
||||
for item in artifacts
|
||||
if isinstance(item, dict) and item.get("path") == asset_path
|
||||
),
|
||||
None,
|
||||
)
|
||||
if descriptor is None:
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB asset not found")
|
||||
relative = PurePosixPath(asset_path)
|
||||
path = candidate.joinpath(*relative.parts)
|
||||
if (
|
||||
relative.is_absolute()
|
||||
or str(relative) != asset_path
|
||||
or any(part in {"", ".", ".."} for part in relative.parts)
|
||||
or path.is_symlink()
|
||||
or not path.is_file()
|
||||
or not path.resolve().is_relative_to(candidate)
|
||||
):
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB asset not found")
|
||||
media_type = descriptor.get("media_type")
|
||||
if not isinstance(media_type, str) or not media_type.startswith("image/"):
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB asset not found")
|
||||
return FileResponse(
|
||||
path,
|
||||
media_type=media_type,
|
||||
headers={
|
||||
"Cache-Control": "private, max-age=31536000, immutable",
|
||||
"ETag": f'"{descriptor.get("sha256", "")}"',
|
||||
"X-Content-Type-Options": "nosniff",
|
||||
},
|
||||
)
|
||||
|
||||
return router
|
||||
|
||||
|
||||
def _configured_root(provider: RootProvider) -> Path | None:
|
||||
candidate = provider()
|
||||
if candidate is None:
|
||||
return None
|
||||
absolute = candidate.expanduser().absolute()
|
||||
if absolute.is_symlink():
|
||||
return None
|
||||
try:
|
||||
root = absolute.resolve(strict=True)
|
||||
except OSError:
|
||||
return None
|
||||
return root if root.is_dir() else None
|
||||
|
||||
|
||||
def _resolve_candidate(provider: RootProvider, result_id: str) -> Path:
|
||||
root = _configured_root(provider)
|
||||
if root is None or RESULT_ID.fullmatch(result_id) is None:
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB result not found")
|
||||
candidate = root / result_id
|
||||
if candidate.is_symlink():
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB result not found")
|
||||
try:
|
||||
resolved = candidate.resolve(strict=True)
|
||||
except OSError:
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB result not found") from None
|
||||
if not resolved.is_dir() or not resolved.is_relative_to(root):
|
||||
raise HTTPException(status_code=404, detail="Vegetation LAB result not found")
|
||||
return resolved
|
||||
|
||||
|
||||
def _read_verified(candidate: Path) -> dict[str, Any]:
|
||||
try:
|
||||
rows: list[tuple[str, int, int, int, int]] = []
|
||||
for path in sorted(candidate.rglob("*"), key=lambda item: item.as_posix()):
|
||||
stat = path.lstat()
|
||||
rows.append((
|
||||
path.relative_to(candidate).as_posix(),
|
||||
stat.st_mode,
|
||||
stat.st_size,
|
||||
stat.st_mtime_ns,
|
||||
stat.st_ctime_ns,
|
||||
))
|
||||
signature = tuple(rows)
|
||||
except OSError:
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail="Vegetation LAB evidence failed verification",
|
||||
) from None
|
||||
return _read_verified_cached(str(candidate), signature)
|
||||
|
||||
|
||||
@lru_cache(maxsize=16)
|
||||
def _read_verified_cached(
|
||||
candidate_text: str,
|
||||
signature: tuple[tuple[str, int, int, int, int], ...],
|
||||
) -> dict[str, Any]:
|
||||
del signature
|
||||
candidate = Path(candidate_text)
|
||||
try:
|
||||
verify_laboratory_evidence_result(_DEFINITION, candidate)
|
||||
path = candidate / "result.json"
|
||||
if path.stat().st_size > _MAX_DOCUMENT_BYTES:
|
||||
raise LaboratoryEvidenceReportError("Vegetation LAB document is too large")
|
||||
payload = json.loads(path.read_text("utf-8"))
|
||||
except (json.JSONDecodeError, OSError, LaboratoryEvidenceReportError):
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail="Vegetation LAB evidence failed verification",
|
||||
) from None
|
||||
if not isinstance(payload, dict):
|
||||
raise HTTPException(status_code=503, detail="Vegetation LAB evidence is invalid")
|
||||
return payload
|
||||
|
||||
|
||||
__all__ = ["build_vegetation_shadow_lab_router"]
|
||||
Reference in New Issue
Block a user