137 lines
5.2 KiB
Python
137 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Pure, inactive title-presence observer contract for Phase 1.0DF."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
import re
|
|
|
|
|
|
PHASE = "PHASE_1_0DF_INACTIVE_TITLE_PRESENCE_OBSERVER"
|
|
FIRMWARE = "9.60"
|
|
TITLE_ID = "PPSA01659"
|
|
MAX_RESULT_BYTES = 4096
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
|
|
|
|
class ObserverError(ValueError):
|
|
"""An observer plan or result is ambiguous or unsafe."""
|
|
|
|
|
|
class Outcome(Enum):
|
|
PRESENT = "PRESENT"
|
|
ABSENT = "ABSENT"
|
|
UNKNOWN = "UNKNOWN"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ObserverPlan:
|
|
phase: str
|
|
active: bool
|
|
firmware: str | None
|
|
title_id: str | None
|
|
method: str | None
|
|
query_contract_sha256: str | None
|
|
exact_literal_path: str | None
|
|
path_provenance_sha256: str | None
|
|
request_bytes_sha256: str | None
|
|
maximum_result_bytes: int | None
|
|
one_request: bool
|
|
read_only: bool
|
|
shell_present: bool
|
|
directory_enumeration: bool
|
|
title_launch: bool
|
|
app_termination: bool
|
|
device_write: bool
|
|
retry: bool
|
|
reconnect: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SyntheticResult:
|
|
firmware: str
|
|
title_id: str
|
|
method: str
|
|
request_bytes_sha256: str
|
|
complete: bool
|
|
result_bytes: int
|
|
explicit_present: bool
|
|
explicit_absent: bool
|
|
error_code: int | None
|
|
|
|
|
|
def validate_inactive(plan: ObserverPlan) -> None:
|
|
"""Require the tracked plan to be empty and incapable of a request."""
|
|
if type(plan) is not ObserverPlan or plan.phase != PHASE or plan.active:
|
|
raise ObserverError("tracked observer is not inactive")
|
|
optional = (plan.firmware, plan.title_id, plan.method,
|
|
plan.query_contract_sha256, plan.exact_literal_path,
|
|
plan.path_provenance_sha256, plan.request_bytes_sha256,
|
|
plan.maximum_result_bytes)
|
|
if any(value is not None for value in optional):
|
|
raise ObserverError("inactive observer contains request data")
|
|
if any((plan.one_request, plan.read_only, plan.shell_present,
|
|
plan.directory_enumeration, plan.title_launch,
|
|
plan.app_termination, plan.device_write, plan.retry,
|
|
plan.reconnect)):
|
|
raise ObserverError("inactive observer contains capability")
|
|
|
|
|
|
def validate_candidate(plan: ObserverPlan) -> None:
|
|
"""Validate hypothetical data without creating or authorizing a request."""
|
|
if type(plan) is not ObserverPlan or plan.phase != PHASE or not plan.active:
|
|
raise ObserverError("candidate is not active data")
|
|
if plan.firmware != FIRMWARE or plan.title_id != TITLE_ID:
|
|
raise ObserverError("firmware or title mismatch")
|
|
if plan.method not in {"SOURCE_BOUND_QUERY", "EXACT_PATH_METADATA"}:
|
|
raise ObserverError("observer method is not allowlisted")
|
|
if not isinstance(plan.query_contract_sha256, str) or \
|
|
not SHA256.fullmatch(plan.query_contract_sha256):
|
|
raise ObserverError("query contract identity is absent")
|
|
if not isinstance(plan.request_bytes_sha256, str) or \
|
|
not SHA256.fullmatch(plan.request_bytes_sha256):
|
|
raise ObserverError("request bytes are not bound")
|
|
if plan.maximum_result_bytes != MAX_RESULT_BYTES:
|
|
raise ObserverError("result bound is not exact")
|
|
if plan.method == "SOURCE_BOUND_QUERY":
|
|
if plan.exact_literal_path is not None or \
|
|
plan.path_provenance_sha256 is not None:
|
|
raise ObserverError("query method must not carry a path")
|
|
else:
|
|
if not isinstance(plan.exact_literal_path, str) or \
|
|
not plan.exact_literal_path.startswith("/") or \
|
|
".." in plan.exact_literal_path or \
|
|
not isinstance(plan.path_provenance_sha256, str) or \
|
|
not SHA256.fullmatch(plan.path_provenance_sha256):
|
|
raise ObserverError("exact path provenance is absent")
|
|
if plan.one_request is not True or plan.read_only is not True:
|
|
raise ObserverError("one read-only request is not exact")
|
|
if any((plan.shell_present, plan.directory_enumeration,
|
|
plan.title_launch, plan.app_termination, plan.device_write,
|
|
plan.retry, plan.reconnect)):
|
|
raise ObserverError("candidate contains a forbidden capability")
|
|
|
|
|
|
def classify_synthetic(plan: ObserverPlan, result: SyntheticResult) -> Outcome:
|
|
"""Classify supplied bytes; errors and incomplete data remain UNKNOWN."""
|
|
validate_candidate(plan)
|
|
if type(result) is not SyntheticResult or \
|
|
(result.firmware, result.title_id, result.method,
|
|
result.request_bytes_sha256) != \
|
|
(plan.firmware, plan.title_id, plan.method,
|
|
plan.request_bytes_sha256):
|
|
raise ObserverError("result binding mismatch")
|
|
if result.result_bytes < 0 or result.result_bytes > MAX_RESULT_BYTES:
|
|
raise ObserverError("result bound exceeded")
|
|
if result.explicit_present and result.explicit_absent:
|
|
raise ObserverError("contradictory result")
|
|
if not result.complete or result.error_code is not None:
|
|
return Outcome.UNKNOWN
|
|
if result.explicit_present:
|
|
return Outcome.PRESENT
|
|
if result.explicit_absent:
|
|
return Outcome.ABSENT
|
|
return Outcome.UNKNOWN
|