tegwick
c96b0c6d88
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
145 lines
7 KiB
Python
145 lines
7 KiB
Python
"""Unexecuted exact readonly provider-derived denial mechanism probe.
|
|
|
|
Run only inside the existing admitted Pod after exact separate founder go.
|
|
Never constructs Store/Runtime/admin or starts delivery/approval/lifecycle effects.
|
|
"""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import sys
|
|
import uuid
|
|
|
|
from informed_decision.http_transport import JSONTransport
|
|
from informed_decision.policy import PolicyClient
|
|
|
|
POLICY_SHA = "9d7c1be4cb6199d14ef0aaead5f85951f45b0ccbd7c800a7fde922a253566b62"
|
|
CONFIG = Path("/run/informed-decision/private/runtime.json")
|
|
PIN = {"origin": "http://flex-auth-informed-decision-lifecycle.flex-auth.svc.cluster.local:8080",
|
|
"package": "informed-decision.lifecycle", "version": "v2",
|
|
"package_digest": "sha256:da01e045fc108adec6b489e5d51f7e57366793016c5a64776963780a6aae457c",
|
|
"caller_token_file": "/var/run/secrets/informed-decision/policy/token"}
|
|
BAD_PURPOSES = ("readback", "enforcement", "unknown", None, False, 0, [], {}, True, " ")
|
|
|
|
|
|
def require(value):
|
|
if not value:
|
|
raise ValueError("provider mechanism proof refused")
|
|
|
|
|
|
def request():
|
|
# This truthful service is NOT the separately authenticated StateHub lifecycle
|
|
# principal. Empty attributes claim no human/group/role/assurance/provenance.
|
|
command = json.dumps({"query": "", "limit": 1}, sort_keys=True, separators=(",", ":"))
|
|
return {"id": str(uuid.uuid4()), "tenant": "tenant:platform",
|
|
"subject": {"id": "system:serviceaccount:informed-decision:review",
|
|
"type": "service", "tenant": "tenant:platform", "attributes": {}},
|
|
"action": "lifecycle.search",
|
|
"resource": {"id": "decisions", "type": "decision-collection",
|
|
"system": "informed-decision", "tenant": "tenant:platform"},
|
|
"context": {"contract": "informed-decision.lifecycle.v1", "memo_version": None,
|
|
"record": {"registered": False},
|
|
"command_digest": "sha256:" + hashlib.sha256(command.encode()).hexdigest()},
|
|
"policy_version": "v2"}
|
|
|
|
|
|
class Capture:
|
|
def __init__(self, real):
|
|
self.real, self.calls, self.captured = real, 0, None
|
|
|
|
def request(self, method, url, *, headers=None, body=None):
|
|
require(self.calls == 0 and method == "POST" and url == PIN["origin"] + "/v1/check")
|
|
self.calls += 1
|
|
status, data = self.real.request(method, url, headers=headers, body=body)
|
|
require(status == 200 and isinstance(data, dict))
|
|
self.captured = copy.deepcopy(data) # Pod memory only; no raw evidence export.
|
|
return status, data
|
|
|
|
|
|
class TransformedCopy:
|
|
def __init__(self, captured, purpose):
|
|
self.captured, self.purpose, self.calls = captured, purpose, 0
|
|
|
|
def request(self, method, url, *, headers=None, body=None):
|
|
# No network/provider call here. This is labelled transformed fixture data.
|
|
self.calls += 1
|
|
value = copy.deepcopy(self.captured)
|
|
value["binding"]["purpose"] = self.purpose
|
|
return 200, value
|
|
|
|
|
|
def prove(token_provider, transport):
|
|
capture = Capture(transport)
|
|
args = {key: PIN[key] for key in ("package", "version", "package_digest")}
|
|
client = PolicyClient(PIN["origin"], token_provider, transport=capture,
|
|
allow_internal_http=True, **args)
|
|
req = request()
|
|
baseline = client.check(req)
|
|
# Expected genuine DENY: this consumer service is not the initial audience.
|
|
# Unavailable/auth failure/malformed response/allow are not successful proof.
|
|
require(capture.calls == 1 and baseline.outcome == "policy_denied"
|
|
and baseline.decision_json is not None and capture.captured is not None)
|
|
require(capture.captured["effect"] == "deny"
|
|
and type(capture.captured["binding"].get("purpose", "")) is str
|
|
and capture.captured["binding"].get("purpose", "") == "")
|
|
before = copy.deepcopy(capture.captured)
|
|
refused = 0
|
|
for purpose in BAD_PURPOSES:
|
|
fixture = TransformedCopy(capture.captured, purpose)
|
|
# No credential needed for these isolated transport copies. PolicyClient
|
|
# requires a nonempty caller string; this value never leaves this function.
|
|
isolated = PolicyClient(PIN["origin"], lambda: "transformed-fixture-only",
|
|
transport=fixture, allow_internal_http=True, **args)
|
|
observed = isolated.check(req)
|
|
require(fixture.calls == 1 and observed.outcome == "invalid_decision"
|
|
and observed.decision_json is None)
|
|
refused += 1
|
|
require(capture.captured == before and capture.calls == 1 and refused == 10)
|
|
return {"kind": "informed-decision.provider-derived-denial-mechanism-proof",
|
|
"all_checks_pass": True, "actual_provider_calls": 1,
|
|
"actual_provider_effect": "deny", "canonical_binding_validated": True,
|
|
"package": PIN["package"], "version": PIN["version"],
|
|
"request_action": "lifecycle.search", "truthful_service_subject": True,
|
|
"human_or_statehub_impersonation": False, "provider_allow_proven": False,
|
|
"transformed_noncanonical_purposes_refused": refused,
|
|
"transformed_fixture_copies_only": True,
|
|
"provider_emitted_advisory_proven": False,
|
|
"database_or_act_effect": False, "credential_or_raw_response_output": False,
|
|
"full_T04_native_lifecycle_acceptance": False}
|
|
|
|
|
|
def main():
|
|
require(not sys.flags.optimize and os.getuid() == 10001 and os.getgid() == 10001)
|
|
import informed_decision.policy as installed
|
|
module = Path(installed.__file__)
|
|
info = module.lstat()
|
|
require(stat.S_ISREG(info.st_mode) and info.st_uid == 0
|
|
and stat.S_IMODE(info.st_mode) == 0o644
|
|
and hashlib.sha256(module.read_bytes()).hexdigest() == POLICY_SHA)
|
|
info = CONFIG.lstat()
|
|
require(stat.S_ISREG(info.st_mode) and info.st_uid == 10001 and info.st_nlink == 1
|
|
and stat.S_IMODE(info.st_mode) == 0o600 and 0 < info.st_size <= 16384)
|
|
config = json.loads(CONFIG.read_text())
|
|
require(config.get("schema") == "informed-decision.review-runtime.v1"
|
|
and config.get("lifecycle_policy") == PIN)
|
|
def token():
|
|
# Reopen existing projected caller custody inside Pod only. Never print,
|
|
# export/decode/cache it or use a human/requester credential instead.
|
|
with Path(PIN["caller_token_file"]).open("rb") as stream:
|
|
raw = stream.read(32769)
|
|
require(0 < len(raw) <= 32768)
|
|
return raw.decode("ascii").strip()
|
|
receipt = prove(token, JSONTransport(allow_internal_http=True))
|
|
print(json.dumps(receipt, sort_keys=True))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except BaseException:
|
|
# Fixed failure only. Raw exception, transport/header/config/body/token
|
|
# are never surfaced. No fallback identity, effect retry or DB write.
|
|
print('{"kind":"informed-decision.provider-derived-denial-mechanism-proof","all_checks_pass":false,"raw_output_withheld":true}')
|
|
raise SystemExit(1) from None
|