feat(identity): add static workspace authority foundation

.gitignore

@@ -17,6 +17,7 @@ env/
 
 # Credentials — NEVER commit these
 *.json
+!opax-mcp/policy/membership_authority.json
 credentials/
 *.key
 *.pem

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

opax-mcp/policy/membership_authority.json

@@ -0,0 +1,5 @@
+{
+  "schema_version": 1,
+  "workspace_id": "ws:vauco",
+  "members": []
+}

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()