| # Copyright 2026 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Tests for the interactive consent flow.""" |
| |
| import json |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from typing import Any |
| |
| from driver_lab.api import EXIT_PERMISSION, DriverLab |
| from driver_lab.consent import ConsentDecision |
| from driver_lab.models import AccessClass, AccessRequest, Decision |
| from driver_lab.permissions import load_grants |
| from driver_lab.transport import FakeProxyTarget, ProxyDescription, ResourceInfo |
| |
| CTRL_DIGEST = "sha256:" + "ab" * 32 |
| DESCRIPTION_DIGEST = "sha256:" + "ee" * 32 |
| POLICY_DIGEST = "sha256:" + "ff" * 32 |
| TARGET_SCOPE = "example-engineering-target" |
| NODE_ID = "example-device" |
| |
| |
| def make_description() -> ProxyDescription: |
| return ProxyDescription( |
| protocol_major=1, |
| protocol_minor=0, |
| proxy_generation=7, |
| boot_id="boot-1", |
| resource_digest=DESCRIPTION_DIGEST, |
| policy_digest=POLICY_DIGEST, |
| resources=( |
| ResourceInfo( |
| id=1, name="control", logical_size=0x100, digest=CTRL_DIGEST |
| ), |
| ), |
| max_snapshot_items=64, |
| audit_capacity=1024, |
| ) |
| |
| |
| def make_plan(**overrides: object) -> dict[str, Any]: |
| plan: dict[str, Any] = { |
| "schema_version": 1, |
| "run_id": "run-1", |
| "case_id": "case-1", |
| "target": {"selector": "lab-target"}, |
| "node": {"id": NODE_ID, "expected_resource_digest": DESCRIPTION_DIGEST}, |
| "access": {"mode": "proxy", "activation": "bind-unclaimed"}, |
| "operations": [ |
| {"kind": "mmio_read32", "resource": "control", "offset": "0x3c"}, |
| ], |
| } |
| plan.update(overrides) |
| return plan |
| |
| |
| class ScriptedPrompt: |
| """Consent prompt returning scripted decisions and recording calls.""" |
| |
| def __init__(self, decision: ConsentDecision) -> None: |
| self.decision = decision |
| self.requests: list[AccessRequest] = [] |
| self.warnings: list[str] = [] |
| |
| async def request_consent( |
| self, request: AccessRequest, warning: str |
| ) -> ConsentDecision: |
| self.requests.append(request) |
| self.warnings.append(warning) |
| return self.decision |
| |
| |
| class ConsentFlowTest(unittest.IsolatedAsyncioTestCase): |
| def setUp(self) -> None: |
| self._dir = tempfile.TemporaryDirectory() |
| self.addCleanup(self._dir.cleanup) |
| base = Path(self._dir.name) |
| self.grants_path = base / "grants.toml" |
| self.evidence_root = base / "evidence" |
| self.fake = FakeProxyTarget(make_description()) |
| self.fake.set_value(1, 0x3C, 0xDEAD_BEEF) |
| |
| def make_lab(self, prompt: ScriptedPrompt | None) -> DriverLab: |
| return DriverLab( |
| self.fake, |
| grants_path=self.grants_path, |
| evidence_root=self.evidence_root, |
| target_scope=TARGET_SCOPE, |
| node_id=NODE_ID, |
| consent=prompt, |
| ) |
| |
| def resolution_sources(self, evidence_dir: Path) -> list[str]: |
| rows = json.loads( |
| (evidence_dir / "permission-resolution.json").read_text() |
| ) |
| return [row["source"] for row in rows] |
| |
| async def test_prompt_receives_exact_rule_and_warning(self) -> None: |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) |
| await self.make_lab(prompt).run_plan(make_plan()) |
| [request] = prompt.requests |
| self.assertEqual(request.resource, "control") |
| self.assertEqual(request.offset, 0x3C) |
| self.assertEqual(request.access, AccessClass.READ_ONCE) |
| self.assertEqual(request.resource_digest, CTRL_DIGEST) |
| self.assertIn("clear status bits", prompt.warnings[0]) |
| |
| async def test_allow_once_succeeds_without_persisting(self) -> None: |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) |
| result = await self.make_lab(prompt).run_plan(make_plan()) |
| self.assertTrue(result.ok, result.failure) |
| self.assertEqual(result.reads[0].value, 0xDEAD_BEEF) |
| self.assertEqual(load_grants(self.grants_path), []) |
| self.assertIn( |
| "allow_once", self.resolution_sources(result.evidence_dir) |
| ) |
| |
| async def test_always_allow_persists_exact_rule(self) -> None: |
| prompt = ScriptedPrompt(ConsentDecision.ALWAYS_ALLOW) |
| result = await self.make_lab(prompt).run_plan(make_plan()) |
| self.assertTrue(result.ok, result.failure) |
| [grant] = load_grants(self.grants_path) |
| self.assertEqual(grant.offset, 0x3C) |
| self.assertEqual(grant.decision, Decision.ALLOW) |
| self.assertEqual(grant.approval_source, "interactive") |
| |
| # A second run resolves from the persistent grant without prompting. |
| second_prompt = ScriptedPrompt(ConsentDecision.DENY_ONCE) |
| second = await self.make_lab(second_prompt).run_plan( |
| make_plan(run_id="run-2") |
| ) |
| self.assertTrue(second.ok, second.failure) |
| self.assertEqual(second_prompt.requests, []) |
| self.assertIn( |
| "persistent", self.resolution_sources(second.evidence_dir) |
| ) |
| |
| async def test_deny_once_fails_without_persisting(self) -> None: |
| prompt = ScriptedPrompt(ConsentDecision.DENY_ONCE) |
| result = await self.make_lab(prompt).run_plan(make_plan()) |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) |
| self.assertEqual(result.failure, "denied by operator") |
| self.assertEqual(self.fake.open_attempts, 0) |
| self.assertEqual(load_grants(self.grants_path), []) |
| |
| async def test_always_deny_persists_and_skips_future_prompts(self) -> None: |
| prompt = ScriptedPrompt(ConsentDecision.ALWAYS_DENY) |
| result = await self.make_lab(prompt).run_plan(make_plan()) |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) |
| [grant] = load_grants(self.grants_path) |
| self.assertEqual(grant.decision, Decision.DENY) |
| |
| second_prompt = ScriptedPrompt(ConsentDecision.ALWAYS_ALLOW) |
| second = await self.make_lab(second_prompt).run_plan( |
| make_plan(run_id="run-2") |
| ) |
| self.assertEqual(second.exit_category, EXIT_PERMISSION) |
| self.assertEqual(second.failure, "denied by persistent grant") |
| self.assertEqual(second_prompt.requests, []) |
| |
| async def test_duplicate_offsets_prompt_once(self) -> None: |
| prompt = ScriptedPrompt(ConsentDecision.ALLOW_ONCE) |
| plan = make_plan( |
| operations=[ |
| { |
| "kind": "mmio_read32", |
| "resource": "control", |
| "offset": "0x3c", |
| }, |
| { |
| "kind": "mmio_read32", |
| "resource": "control", |
| "offset": "0x3c", |
| }, |
| ] |
| ) |
| result = await self.make_lab(prompt).run_plan(plan) |
| self.assertTrue(result.ok, result.failure) |
| self.assertEqual(len(prompt.requests), 1) |
| |
| async def test_no_prompt_fails_closed(self) -> None: |
| result = await self.make_lab(None).run_plan(make_plan()) |
| self.assertEqual(result.exit_category, EXIT_PERMISSION) |
| self.assertEqual( |
| result.failure, |
| "consent required; unattended operation fails closed", |
| ) |
| self.assertIn( |
| "fail_closed", self.resolution_sources(result.evidence_dir) |
| ) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
