326 lines
13 KiB
Python
326 lines
13 KiB
Python
"""Tests for the V2R1C6 governed local writer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import struct
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from datetime import datetime, timezone
|
|
|
|
import governed_local_writer as gw
|
|
from aios_governed_local_writer_service import GovernedLocalWriterService, InMemoryAuditSink
|
|
|
|
|
|
class FakeHandler:
|
|
"""Records calls from service for assertions."""
|
|
|
|
def __init__(self) -> None:
|
|
self.create_calls: list[tuple] = []
|
|
self.append_calls: list[tuple] = []
|
|
|
|
def create_branch(self, request: gw.CreateBranchRequest) -> dict[str, str]:
|
|
self.create_calls.append((request.branch, request.base_oid, request.head_oid))
|
|
return {"status": "ok"}
|
|
|
|
def append_commit(self, request: gw.AppendCommitRequest) -> dict[str, str]:
|
|
self.append_calls.append((request.branch, request.expected_head_oid, tuple(request.changes)))
|
|
return {"status": "ok"}
|
|
|
|
|
|
class FakeConnection:
|
|
"""Stub socket-like object used by peer identity tests."""
|
|
|
|
def __init__(self, pid: int, uid: int, gid: int) -> None:
|
|
self.pid = pid
|
|
self.uid = uid
|
|
self.gid = gid
|
|
|
|
def getsockopt(self, *_args: object) -> bytes:
|
|
return struct.pack("3i", self.pid, self.uid, self.gid)
|
|
|
|
|
|
class GovernedLocalWriterTests(unittest.TestCase):
|
|
"""Coverage for authorization parsing and service semantics."""
|
|
|
|
def _make_auth_payload(
|
|
self,
|
|
*,
|
|
branch: str = "candidate",
|
|
base_oid: str = "0" * 40,
|
|
head_oid: str = "1" * 40,
|
|
expected_head_oid: str = "1" * 40,
|
|
allowed_paths: list[str] | tuple[str, ...] | None = None,
|
|
denied_operations: list[str] | None = None,
|
|
) -> dict[str, object]:
|
|
now = datetime.now(tz=timezone.utc)
|
|
payload: dict[str, object] = {
|
|
"task_id": gw.TASK_ID,
|
|
"repo": gw.REPO,
|
|
"created_at": now.isoformat().replace("+00:00", "Z"),
|
|
"one_time_use": True,
|
|
"allowed_branch_operations": {
|
|
branch: {
|
|
"create_branch": {
|
|
"base_oid": base_oid,
|
|
"head_oid": head_oid,
|
|
},
|
|
"append_commit": {
|
|
"expected_head_oid": expected_head_oid,
|
|
"allowed_paths": list(allowed_paths or ["allowed/file.txt"]),
|
|
},
|
|
}
|
|
},
|
|
"denied_operations": denied_operations or [],
|
|
}
|
|
payload["authorization_sha256"] = gw.compute_authorization_sha256(payload)
|
|
return payload
|
|
|
|
def _write_auth_file(self, payload: dict[str, object], directory: Path) -> Path:
|
|
auth_path = directory / "auth.json"
|
|
auth_path.write_text(json.dumps(payload), encoding="utf-8")
|
|
os.chmod(auth_path, 0o600)
|
|
return auth_path
|
|
|
|
def _load_auth(
|
|
self, payload: dict[str, object], directory: Path, *,
|
|
now: datetime | None = None
|
|
) -> gw.GovernedAuthorization:
|
|
auth_path = self._write_auth_file(payload, directory)
|
|
if now is None:
|
|
now = datetime.now(tz=timezone.utc)
|
|
with patch("governed_local_writer._assert_authorization_file_permissions"):
|
|
return gw.load_authorization_file(auth_path, now=now)
|
|
|
|
def test_authorization_file_rejection(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
directory = Path(root)
|
|
payload = self._make_auth_payload()
|
|
auth_path = self._write_auth_file(payload, directory)
|
|
os.chmod(auth_path, 0o644)
|
|
self.assertTrue(auth_path.exists())
|
|
with self.assertRaises(gw.AuthorizationError):
|
|
gw.load_authorization_file(auth_path)
|
|
|
|
def test_authorization_canonical_hash(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
directory = Path(root)
|
|
payload = self._make_auth_payload()
|
|
payload["authorization_sha256"] = "0" * 64
|
|
auth_path = self._write_auth_file(payload, directory)
|
|
with patch("governed_local_writer._assert_authorization_file_permissions"):
|
|
with self.assertRaises(gw.AuthorizationError):
|
|
gw.load_authorization_file(auth_path)
|
|
|
|
def test_peer_identity_helper(self) -> None:
|
|
if os.name == "nt" or not hasattr(gw.socket, "SO_PEERCRED"):
|
|
self.skipTest("SO_PEERCRED not available on this platform")
|
|
|
|
connection = FakeConnection(pid=111, uid=0, gid=222)
|
|
with patch("governed_local_writer._resolve_peer_executable", return_value="/usr/bin/ci-worker"):
|
|
identity = gw.extract_peer_identity(connection)
|
|
self.assertEqual(identity.pid, 111)
|
|
self.assertEqual(identity.uid, 0)
|
|
self.assertEqual(identity.executable, "/usr/bin/ci-worker")
|
|
|
|
forbidden = FakeConnection(pid=111, uid=1000, gid=222)
|
|
with patch("governed_local_writer._resolve_peer_executable", return_value="/usr/bin/ci-worker"):
|
|
with self.assertRaises(PermissionError):
|
|
gw.extract_peer_identity(forbidden)
|
|
|
|
def test_branch_policy_enforced(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
directory = Path(root)
|
|
auth = self._load_auth(self._make_auth_payload(), directory)
|
|
handler = FakeHandler()
|
|
service = GovernedLocalWriterService(
|
|
Path(directory) / "socket",
|
|
auth,
|
|
handler,
|
|
audit_sink=InMemoryAuditSink(),
|
|
)
|
|
peer = gw.PeerIdentity(1, 0, 1, None)
|
|
|
|
create_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "create_branch",
|
|
"branch": "candidate",
|
|
"base_oid": "0" * 40,
|
|
"head_oid": "1" * 40,
|
|
}
|
|
create_response = service.process_request(create_request, peer)
|
|
self.assertEqual(create_response["status"], "ok")
|
|
|
|
bad_request = dict(create_request)
|
|
bad_request["branch"] = "forbidden"
|
|
bad_response = service.process_request(bad_request, peer)
|
|
self.assertEqual(bad_response["status"], "error")
|
|
self.assertEqual(bad_response["operation"], "create_branch")
|
|
self.assertEqual(bad_response["branch"], "forbidden")
|
|
self.assertEqual(len(service.audit_sink.records), 2)
|
|
|
|
def test_path_policy(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
directory = Path(root)
|
|
auth = self._load_auth(
|
|
self._make_auth_payload(
|
|
allowed_paths=("allowed/dir",),
|
|
),
|
|
directory,
|
|
)
|
|
handler = FakeHandler()
|
|
service = GovernedLocalWriterService(
|
|
Path(directory) / "socket",
|
|
auth,
|
|
handler,
|
|
audit_sink=InMemoryAuditSink(),
|
|
)
|
|
peer = gw.PeerIdentity(2, 0, 2, None)
|
|
|
|
create_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "create_branch",
|
|
"branch": "candidate",
|
|
"base_oid": "0" * 40,
|
|
"head_oid": "1" * 40,
|
|
}
|
|
self.assertEqual(service.process_request(create_request, peer)["status"], "ok")
|
|
|
|
blocked_path_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "append_commit",
|
|
"branch": "candidate",
|
|
"expected_head_oid": "1" * 40,
|
|
"changes": [
|
|
{"path": "blocked.txt", "content": "hello"},
|
|
],
|
|
}
|
|
self.assertEqual(service.process_request(blocked_path_request, peer)["status"], "error")
|
|
|
|
blocked_segment_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "append_commit",
|
|
"branch": "candidate",
|
|
"expected_head_oid": "1" * 40,
|
|
"changes": [
|
|
{"path": "allowed/dir/secret.txt", "content": "hello"},
|
|
],
|
|
}
|
|
self.assertEqual(service.process_request(blocked_segment_request, peer)["status"], "error")
|
|
|
|
def test_secret_content_scan(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
directory = Path(root)
|
|
auth = self._load_auth(self._make_auth_payload(), directory)
|
|
handler = FakeHandler()
|
|
service = GovernedLocalWriterService(
|
|
Path(directory) / "socket",
|
|
auth,
|
|
handler,
|
|
audit_sink=InMemoryAuditSink(),
|
|
)
|
|
peer = gw.PeerIdentity(3, 0, 3, None)
|
|
|
|
create_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "create_branch",
|
|
"branch": "candidate",
|
|
"base_oid": "0" * 40,
|
|
"head_oid": "1" * 40,
|
|
}
|
|
self.assertEqual(service.process_request(create_request, peer)["status"], "ok")
|
|
|
|
secret_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "append_commit",
|
|
"branch": "candidate",
|
|
"expected_head_oid": "1" * 40,
|
|
"changes": [
|
|
{"path": "allowed/file.txt", "content": "pass" + "word=" + "x" * 24},
|
|
],
|
|
}
|
|
self.assertEqual(service.process_request(secret_request, peer)["status"], "error")
|
|
|
|
def test_one_shot_consumption(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
directory = Path(root)
|
|
auth = self._load_auth(self._make_auth_payload(), directory)
|
|
handler = FakeHandler()
|
|
sink = InMemoryAuditSink()
|
|
service = GovernedLocalWriterService(
|
|
Path(directory) / "socket",
|
|
auth,
|
|
handler,
|
|
audit_sink=sink,
|
|
)
|
|
peer = gw.PeerIdentity(4, 0, 4, None)
|
|
|
|
create_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "create_branch",
|
|
"branch": "candidate",
|
|
"base_oid": "0" * 40,
|
|
"head_oid": "1" * 40,
|
|
}
|
|
append_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "append_commit",
|
|
"branch": "candidate",
|
|
"expected_head_oid": "1" * 40,
|
|
"changes": [{"path": "allowed/file.txt", "content": "safe line"}],
|
|
}
|
|
|
|
self.assertEqual(service.process_request(create_request, peer)["status"], "ok")
|
|
self.assertEqual(service.process_request(append_request, peer)["status"], "ok")
|
|
self.assertEqual(service.process_request(append_request, peer)["status"], "error")
|
|
self.assertEqual(len(sink.records), 3)
|
|
self.assertEqual(len(handler.create_calls), 1)
|
|
self.assertEqual(len(handler.append_calls), 1)
|
|
|
|
def test_response_is_sanitized(self) -> None:
|
|
with tempfile.TemporaryDirectory() as root:
|
|
directory = Path(root)
|
|
auth = self._load_auth(self._make_auth_payload(), directory)
|
|
handler = FakeHandler()
|
|
service = GovernedLocalWriterService(
|
|
Path(directory) / "socket",
|
|
auth,
|
|
handler,
|
|
audit_sink=InMemoryAuditSink(),
|
|
)
|
|
peer = gw.PeerIdentity(5, 0, 5, None)
|
|
|
|
bad_request = {
|
|
"task_id": gw.TASK_ID,
|
|
"principal": gw.PRINCIPAL,
|
|
"op": "append_commit",
|
|
"branch": "candidate",
|
|
"expected_head_oid": "deadbeef" * 5,
|
|
"changes": [
|
|
{"path": "allowed/file.txt", "content": "pass" + "word=" + "x" * 24},
|
|
],
|
|
"authorization": "super-secret-token",
|
|
}
|
|
response = service.process_request(bad_request, peer)
|
|
text = json.dumps(response)
|
|
self.assertEqual(response["status"], "error")
|
|
self.assertNotIn("authorization", response)
|
|
self.assertNotIn("super-secret-token", text)
|
|
self.assertNotIn("pass" + "word=" + "x" * 24, text)
|
|
self.assertEqual(set(response.keys()), {"status", "operation", "branch", "error"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|