From 8aa533eecc33ab4dc449d83e1c11304b60c919cd Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Sat, 19 Sep 2026 17:18:38 +0000 Subject: [PATCH] feat(identity): add static workspace authority foundation --- .gitignore | 1 + opax-mcp/policy/authority.py | 153 +++++++++++++ opax-mcp/policy/membership_authority.json | 5 + opax-mcp/test_membership_authority.py | 267 ++++++++++++++++++++++ 4 files changed, 426 insertions(+) create mode 100644 opax-mcp/policy/authority.py create mode 100644 opax-mcp/policy/membership_authority.json create mode 100644 opax-mcp/test_membership_authority.py diff --git a/.gitignore b/.gitignore index a609430..6e6b224 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ env/ # Credentials — NEVER commit these *.json +!opax-mcp/policy/membership_authority.json credentials/ *.key *.pem diff --git a/opax-mcp/policy/authority.py b/opax-mcp/policy/authority.py new file mode 100644 index 0000000..d20f347 --- /dev/null +++ b/opax-mcp/policy/authority.py @@ -0,0 +1,153 @@ +""" +Provides a static, file-based authority for workspace membership. + +This module loads and validates a JSON file defining members of a single workspace. +It is intended for an initial, internal-only launch and does not provide +multi-workspace support. +""" +import json +import pathlib +from dataclasses import dataclass +from types import MappingProxyType +from typing import Dict, List, Literal, Mapping, Optional + +# --- Public API --- + +class MembershipAuthorityError(Exception): + """Custom exception for errors related to the membership authority.""" + pass + +@dataclass(frozen=True) +class MemberContext: + """Represents the validated context of an authorized member.""" + principal_id: str + workspace_id: str + role: Literal["admin", "operator", "member"] + status: Literal["active", "suspended", "revoked"] + +# --- Internal Implementation --- + +@dataclass(frozen=True) +class _MembershipAuthority: + """Represents the entire validated membership authority configuration.""" + schema_version: int + workspace_id: str + members: Mapping[str, MemberContext] + +def _validate_authority_data(data: dict) -> _MembershipAuthority: + """Performs strict validation of the raw data from JSON.""" + _EXPECTED_TOP_LEVEL_KEYS = {"schema_version", "workspace_id", "members"} + _EXPECTED_MEMBER_KEYS = {"principal_id", "workspace_id", "role", "status"} + _VALID_ROLES = {"admin", "operator", "member"} + _VALID_STATUSES = {"active", "suspended", "revoked"} + + if not isinstance(data, dict) or set(data.keys()) != _EXPECTED_TOP_LEVEL_KEYS: + raise MembershipAuthorityError(f"Authority data must be a JSON object with exact keys: {_EXPECTED_TOP_LEVEL_KEYS}") + + schema_version = data["schema_version"] + if type(schema_version) is not int or schema_version != 1: + raise MembershipAuthorityError(f"Unsupported schema_version: must be integer 1.") + + workspace_id = data.get("workspace_id") + if not isinstance(workspace_id, str) or not workspace_id.strip(): + raise MembershipAuthorityError("'workspace_id' must be a non-empty string.") + + if not isinstance(data.get("members"), list): + raise MembershipAuthorityError("'members' must be a list.") + + members_map: Dict[str, MemberContext] = {} + for member_data in data["members"]: + if not isinstance(member_data, dict) or set(member_data.keys()) != _EXPECTED_MEMBER_KEYS: + raise MembershipAuthorityError(f"Member data must be an object with exact keys: {_EXPECTED_MEMBER_KEYS}") + + principal_id = member_data.get("principal_id") + prefix = "user:google:" + if (not isinstance(principal_id, str) or not principal_id.strip() or + not principal_id.startswith(prefix)): + raise MembershipAuthorityError(f"Invalid principal_id format for '{principal_id}'.") + + suffix = principal_id[len(prefix):] + if not suffix.strip(): + raise MembershipAuthorityError(f"Invalid principal_id format for '{principal_id}': empty suffix.") + + if principal_id in members_map: + raise MembershipAuthorityError(f"Duplicate principal_id found: {principal_id}") + if member_data.get("workspace_id") != workspace_id: + raise MembershipAuthorityError(f"Member workspace_id mismatch for {principal_id}.") + + role = member_data.get("role") + if not isinstance(role, str) or role not in _VALID_ROLES: + raise MembershipAuthorityError(f"Invalid role for {principal_id}: {role}") + + status = member_data.get("status") + if not isinstance(status, str) or status not in _VALID_STATUSES: + raise MembershipAuthorityError(f"Invalid status for {principal_id}: {status}") + members_map[principal_id] = MemberContext(**member_data) + + return _MembershipAuthority( + schema_version=schema_version, + workspace_id=workspace_id, + members=MappingProxyType(members_map), + ) + +# --- Public API Functions --- + +def load_membership_authority( + authority_path: Optional[pathlib.Path] = None, +) -> _MembershipAuthority: + """ + Loads and validates the membership_authority.json file. + + Args: + authority_path: Optional path to the authority file for testing. + + Returns: + A validated, internal _MembershipAuthority object. + + Raises: + MembershipAuthorityError: If the file is not found, malformed, or fails validation. + """ + if authority_path is None: + authority_path = pathlib.Path(__file__).parent / "membership_authority.json" + + try: + with open(authority_path, "r", encoding="utf-8") as f: + raw_data = json.load(f) + except FileNotFoundError as e: + raise MembershipAuthorityError(f"Authority file not found at {authority_path}") from e + except OSError as e: + raise MembershipAuthorityError(f"Authority file could not be read at {authority_path}") from e + except UnicodeDecodeError as e: + raise MembershipAuthorityError(f"Authority file is not valid UTF-8 at {authority_path}") from e + except json.JSONDecodeError as e: + raise MembershipAuthorityError(f"Authority file is not valid JSON: {e}") from e + + return _validate_authority_data(raw_data) + +def get_member_context_by_principal( + principal_id: str, authority: Optional[_MembershipAuthority] = None +) -> Optional[MemberContext]: + """ + Retrieves the context for a given principal_id. + + If authority is not provided, it loads from the default module-relative file. + An invalid authority file will raise MembershipAuthorityError. + + Returns a MemberContext only if the principal exists and their status is 'active'. + Returns None for unknown, suspended, or revoked principals. + + Args: + principal_id: The canonical principal ID to look up. + authority: Optional pre-loaded authority object for testing. + + Returns: + A MemberContext object or None. + """ + auth_to_check = authority + if auth_to_check is None: + auth_to_check = load_membership_authority() + + member = auth_to_check.members.get(principal_id) + if member and member.status == "active": + return member + return None diff --git a/opax-mcp/policy/membership_authority.json b/opax-mcp/policy/membership_authority.json new file mode 100644 index 0000000..818a0fc --- /dev/null +++ b/opax-mcp/policy/membership_authority.json @@ -0,0 +1,5 @@ +{ + "schema_version": 1, + "workspace_id": "ws:vauco", + "members": [] +} diff --git a/opax-mcp/test_membership_authority.py b/opax-mcp/test_membership_authority.py new file mode 100644 index 0000000..d596f4f --- /dev/null +++ b/opax-mcp/test_membership_authority.py @@ -0,0 +1,267 @@ +import ast +import json +import pathlib +import sys +import tempfile +import unittest +from unittest.mock import patch + +# Ensure the module under test is on the path +REPO_ROOT = pathlib.Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) + +from policy import authority + +class TestMembershipAuthority(unittest.TestCase): + + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.temp_path = pathlib.Path(self.temp_dir.name) + self.valid_member_active = { + "principal_id": "user:google:1001", + "workspace_id": "ws:vauco", + "role": "member", + "status": "active", + } + + def tearDown(self): + self.temp_dir.cleanup() + + def _write_authority_file(self, data: dict): + path = self.temp_path / "authority.json" + with open(path, "w", encoding="utf-8") as f: + json.dump(data, f) + return path + + # Test Cases + + def test_01_empty_valid_authority_loads(self): + """1. Empty valid authority loads successfully.""" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": []}) + auth = authority.load_membership_authority(path) + self.assertEqual(auth.schema_version, 1) + self.assertEqual(auth.workspace_id, "ws:vauco") + self.assertEqual(len(auth.members), 0) + + def test_02_active_member_lookup(self): + """2. Active member lookup returns exact principal_id, workspace_id, role, status.""" + path = self._write_authority_file({ + "schema_version": 1, "workspace_id": "ws:vauco", "members": [self.valid_member_active] + }) + auth = authority.load_membership_authority(path) + member = authority.get_member_context_by_principal("user:google:1001", auth) + self.assertIsInstance(member, authority.MemberContext) + self.assertEqual(member.principal_id, "user:google:1001") + self.assertEqual(member.workspace_id, "ws:vauco") + self.assertEqual(member.role, "member") + self.assertEqual(member.status, "active") + + def test_03_unknown_member_returns_none(self): + """3. Unknown member returns None.""" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": []}) + auth = authority.load_membership_authority(path) + self.assertIsNone(authority.get_member_context_by_principal("user:google:9999", auth)) + + def test_04_suspended_member_returns_none(self): + """4. Suspended member returns None.""" + member = self.valid_member_active.copy() + member["status"] = "suspended" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + auth = authority.load_membership_authority(path) + self.assertIsNone(authority.get_member_context_by_principal("user:google:1001", auth)) + + def test_05_revoked_member_returns_none(self): + """5. Revoked member returns None.""" + member = self.valid_member_active.copy() + member["status"] = "revoked" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + auth = authority.load_membership_authority(path) + self.assertIsNone(authority.get_member_context_by_principal("user:google:1001", auth)) + + def test_06_duplicate_principal_fails(self): + """6. Duplicate principal IDs fail closed.""" + path = self._write_authority_file({ + "schema_version": 1, "workspace_id": "ws:vauco", "members": [self.valid_member_active, self.valid_member_active] + }) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_07_schema_version_as_true_fails(self): + """7. schema_version=True fails closed.""" + path = self._write_authority_file({"schema_version": True, "workspace_id": "ws:vauco", "members": []}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + def test_08_schema_version_as_float_fails(self): + """8. schema_version=1.0 fails closed.""" + path = self._write_authority_file({"schema_version": 1.0, "workspace_id": "ws:vauco", "members": []}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_09_unsupported_int_schema_version_fails(self): + """9. Unsupported integer schema version fails closed.""" + path = self._write_authority_file({"schema_version": 2, "workspace_id": "ws:vauco", "members": []}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_10_missing_members_key_fails(self): + """10. Missing members fails closed.""" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco"}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_11_unexpected_top_level_key_fails(self): + """11. Unexpected top-level key fails closed.""" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [], "extra": True}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_12_unexpected_member_key_fails(self): + """12. Unexpected member key fails closed.""" + member = self.valid_member_active.copy() + member["extra"] = True + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_13_bad_principal_prefix_fails(self): + """13. user:internal:1001 fails closed.""" + member = self.valid_member_active.copy() + member["principal_id"] = "user:internal:1001" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_14_empty_principal_suffix_fails(self): + """14. user:google: fails closed.""" + member = self.valid_member_active.copy() + member["principal_id"] = "user:google:" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_15_whitespace_workspace_id_fails(self): + """15. Whitespace-only workspace_id fails closed.""" + path = self._write_authority_file({"schema_version": 1, "workspace_id": " ", "members": []}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_16_member_workspace_mismatch_fails(self): + """16. Member workspace mismatch fails closed.""" + member = self.valid_member_active.copy() + member["workspace_id"] = "ws:other" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_17_invalid_role_string_fails(self): + """17. Invalid role fails closed.""" + member = self.valid_member_active.copy() + member["role"] = "superuser" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_18_non_string_role_fails(self): + """18. Non-string role fails closed.""" + member = self.valid_member_active.copy() + member["role"] = ["admin"] + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_19_invalid_status_string_fails(self): + """19. Invalid status fails closed.""" + member = self.valid_member_active.copy() + member["status"] = "pending" + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_20_non_string_status_fails(self): + """20. Non-string status fails closed.""" + member = self.valid_member_active.copy() + member["status"] = 1 + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_21_default_module_loading_succeeds(self): + """21. Default module-relative loading succeeds and verifies content.""" + try: + auth = authority.load_membership_authority() + self.assertEqual(auth.schema_version, 1) + self.assertEqual(auth.workspace_id, "ws:vauco") + self.assertEqual(len(auth.members), 0) + except authority.MembershipAuthorityError as e: + self.fail(f"Default loading failed: {e}") + + def test_22_get_member_propagates_load_error(self): + """22. get_member_context_by_principal propagates load errors.""" + with patch("policy.authority.load_membership_authority") as mock_load: + mock_load.side_effect = authority.MembershipAuthorityError("Config is broken") + with self.assertRaises(authority.MembershipAuthorityError): + authority.get_member_context_by_principal("user:google:1001") + + def test_23_ast_check_for_std_lib_only(self): + """23. AST check confirms authority.py imports only standard-library modules.""" + module_path = REPO_ROOT / "opax-mcp" / "policy" / "authority.py" + with open(module_path, "r", encoding="utf-8") as f: + tree = ast.parse(f.read()) + allowed_imports = {"json", "pathlib", "dataclasses", "typing", "types"} + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + self.assertIn(alias.name.split('.')[0], allowed_imports) + elif isinstance(node, ast.ImportFrom): + if node.module: + # Allow local imports within the same package + if '.' not in node.module: + self.assertIn(node.module.split('.')[0], allowed_imports) + + def test_24_no_at_character_in_test_file(self): + """24. Confirms the test file contains no at-sign character.""" + module_path = pathlib.Path(__file__) + content = module_path.read_text(encoding="utf-8") + self.assertNotIn(chr(64), content) + + def test_25_ast_check_for_forbidden_test_imports(self): + """25. AST check confirms test does not import forbidden modules.""" + module_path = pathlib.Path(__file__) + with open(module_path, "r", encoding="utf-8") as f: + tree = ast.parse(f.read()) + forbidden_imports = { + "server", "google", "firebase_admin", "firestore", "httpx", "requests", "os", "pydantic" + } + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + self.assertNotIn(alias.name.split('.')[0], forbidden_imports) + elif isinstance(node, ast.ImportFrom): + if node.module: + self.assertNotIn(node.module.split('.')[0], forbidden_imports) + + def test_whitespace_only_suffix_fails(self): + """Tests that a principal with a whitespace-only suffix fails.""" + member = self.valid_member_active.copy() + member["principal_id"] = "user:google: " + path = self._write_authority_file({"schema_version": 1, "workspace_id": "ws:vauco", "members": [member]}) + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + def test_unreadable_file_fails(self): + """Tests that an unreadable file raises MembershipAuthorityError.""" + path = self.temp_path / "unreadable_authority.json" + with patch("builtins.open", side_effect=OSError("Permission denied")): + with self.assertRaises(authority.MembershipAuthorityError): + authority.load_membership_authority(path) + + def test_immutable_mapping(self): + """Tests that the loaded members mapping is immutable.""" + path = self._write_authority_file({ + "schema_version": 1, "workspace_id": "ws:vauco", "members": [self.valid_member_active] + }) + auth = authority.load_membership_authority(path) + with self.assertRaises(TypeError): + auth.members["new_user"] = "test" + +if __name__ == "__main__": + unittest.main()