From a777e4abaf998903cfcdd4ae92e79fb14c1ecf0b Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Wed, 16 Sep 2026 16:22:32 +0000 Subject: [PATCH] feat(opax-mcp): add master hub contracts and policy foundation --- opax-mcp/contracts/__init__.py | 1 + opax-mcp/contracts/common.py | 123 +++++++++++++++++++++++++++ opax-mcp/persistence/__init__.py | 1 + opax-mcp/persistence/memory_store.py | 61 +++++++++++++ opax-mcp/policy/__init__.py | 1 + opax-mcp/policy/caller_context.py | 66 ++++++++++++++ opax-mcp/policy/tool_policy.py | 90 ++++++++++++++++++++ opax-mcp/test_phase2a_foundation.py | 119 ++++++++++++++++++++++++++ 8 files changed, 462 insertions(+) create mode 100644 opax-mcp/contracts/__init__.py create mode 100644 opax-mcp/contracts/common.py create mode 100644 opax-mcp/persistence/__init__.py create mode 100644 opax-mcp/persistence/memory_store.py create mode 100644 opax-mcp/policy/__init__.py create mode 100644 opax-mcp/policy/caller_context.py create mode 100644 opax-mcp/policy/tool_policy.py create mode 100644 opax-mcp/test_phase2a_foundation.py diff --git a/opax-mcp/contracts/__init__.py b/opax-mcp/contracts/__init__.py new file mode 100644 index 0000000..c22e35f --- /dev/null +++ b/opax-mcp/contracts/__init__.py @@ -0,0 +1 @@ +# This file makes the 'contracts' directory a Python package. diff --git a/opax-mcp/contracts/common.py b/opax-mcp/contracts/common.py new file mode 100644 index 0000000..fb7c6c1 --- /dev/null +++ b/opax-mcp/contracts/common.py @@ -0,0 +1,123 @@ +""" +Defines the core Pydantic data models (contracts) for the Emma Master Hub. + +These models ensure data consistency and validation across services. +""" +import uuid +from datetime import datetime, timezone +from typing import List, Dict, Any, Optional, Literal +from pydantic import BaseModel, Field + +# --- Core Data Types --- + +SchemaVersion = Literal["1.0.0"] +CallerType = Literal["human", "agent", "system"] +Classification = Literal["public", "internal", "confidential", "secret"] +MemoryType = Literal["fact", "pattern", "procedure", "preference", "chat_summary", "ticket_ref"] +ToolRiskLevel = Literal["read_only", "propose_only", "requires_approval", "requires_high_approval", "forbidden"] +ApprovalStatus = Literal["PENDING", "APPROVED", 'REJECTED', "EXPIRED"] +ExecutionStatus = Literal["PENDING", "EXECUTING", "SUCCESS", "FAILED"] + +# --- Context and Identity Contracts --- + +class CallerContext(BaseModel): + """Server-derived context about the authenticated caller.""" + caller_id: str # e.g., "agent:perplexity" or "user:chris.c" + caller_type: CallerType + profile: str # e.g., "operator", "viewer", "admin" + owner_id: str # The user or service account owning the session + workspace_id: str + allowed_tool_policy: Dict[str, ToolRiskLevel] = Field(default_factory=dict) + schema_version: SchemaVersion = "1.0.0" + +# --- Core Object Contracts --- + +class EmmaConversation(BaseModel): + """Metadata for a single conversation session.""" + conversation_id: str = Field(default_factory=lambda: f"convo-{uuid.uuid4().hex}") + owner_id: str + workspace_id: str + task_id: Optional[str] = None + created_by: str + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + last_updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + classification: Classification = "internal" + # Raw messages are stored separately, this holds a reference or summary + message_references: List[str] = Field(default_factory=list) + schema_version: SchemaVersion = "1.0.0" + +class MemoryRecord(BaseModel): + """A single, recallable piece of information for Emma.""" + memory_id: str = Field(default_factory=lambda: f"mem-{uuid.uuid4().hex}") + memory_type: MemoryType + owner_id: str + workspace_id: str + source_conversation_id: str + content_text: str + embedding_vector_ref: Optional[str] = None + created_by: str + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + last_accessed_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + access_count: int = 1 + reward_score: float = 0.0 + classification: Classification = "internal" + schema_version: SchemaVersion = "1.0.0" + +class ProposedToolAction(BaseModel): + """A tool action proposed by Emma, awaiting approval.""" + action_id: str = Field(default_factory=lambda: f"act-{uuid.uuid4().hex}") + correlation_id: str + tool_name: str + parameters: Dict[str, Any] + human_readable_summary: str + risk_class: ToolRiskLevel + target_resource: Optional[str] = None + source_revision_sha: Optional[str] = None + requires_approval: bool + schema_version: SchemaVersion = "1.0.0" + +class ApprovalRecord(BaseModel): + """A record of an approval process for a ProposedToolAction.""" + approval_id: str = Field(default_factory=lambda: f"appr-{uuid.uuid4().hex}") + action: ProposedToolAction + status: ApprovalStatus = "PENDING" + execution_status: ExecutionStatus = "PENDING" + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + expires_at: datetime + requested_by_caller_id: str + approved_by_user_id: Optional[str] = None + actioned_at: Optional[datetime] = None + execution_log: List[str] = Field(default_factory=list) + idempotency_key: str = Field(default_factory=lambda: uuid.uuid4().hex) + workspace_id: str + schema_version: SchemaVersion = "1.0.0" + +class AuditEvent(BaseModel): + """A discrete, immutable event for audit purposes.""" + event_id: str = Field(default_factory=lambda: f"aud-{uuid.uuid4().hex}") + timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + event_type: str # e.g., "TOOL_PROPOSED", "APPROVAL_GRANTED", "TOOL_EXECUTED" + caller_id: str + user_id: Optional[str] = None + details: Dict[str, Any] + workspace_id: str + schema_version: SchemaVersion = "1.0.0" + +# --- API Contracts for call_emma --- + +class CallEmmaRequest(BaseModel): + prompt: str + conversation_id: Optional[str] = None + task_id: Optional[str] = None + requested_memory_scope: List[MemoryType] = Field(default_factory=list) + client_context: Dict[str, Any] = Field(default_factory=dict) + +class CallEmmaResponse(BaseModel): + reply_text: str + model: str + conversation_id: str + correlation_id: str + memory_references: List[str] = Field(default_factory=list) + proposed_actions: List[ProposedToolAction] = Field(default_factory=list) + approval_state: Optional[Dict[str, Any]] = None + schema_version: SchemaVersion = "1.0.0" diff --git a/opax-mcp/persistence/__init__.py b/opax-mcp/persistence/__init__.py new file mode 100644 index 0000000..a849417 --- /dev/null +++ b/opax-mcp/persistence/__init__.py @@ -0,0 +1 @@ +# This file makes the 'persistence' directory a Python package. diff --git a/opax-mcp/persistence/memory_store.py b/opax-mcp/persistence/memory_store.py new file mode 100644 index 0000000..205ba10 --- /dev/null +++ b/opax-mcp/persistence/memory_store.py @@ -0,0 +1,61 @@ +""" +Defines the abstract interface for Emma's memory systems. +""" +from abc import ABC, abstractmethod +from typing import List, Dict, Any, Optional + +from contracts.common import EmmaConversation, MemoryRecord + + +class MemoryStore(ABC): + """Abstract Base Class for all memory store implementations.""" + + @abstractmethod + async def get_conversation(self, conversation_id: str, owner_id: str) -> Optional[EmmaConversation]: + pass + + @abstractmethod + async def save_conversation(self, conversation: EmmaConversation) -> None: + pass + + @abstractmethod + async def store_memory(self, record: MemoryRecord) -> None: + pass + + @abstractmethod + async def recall_memory(self, embedding: List[float], owner_id: str, scope: List[str]) -> List[MemoryRecord]: + pass + + +class InMemoryMemoryStore(MemoryStore): + """ + A non-persistent, in-memory implementation of the MemoryStore for testing and + local development. DO NOT USE IN PRODUCTION. + """ + def __init__(self): + self._conversations: Dict[str, EmmaConversation] = {} + self._memories: List[MemoryRecord] = [] + print("WARNING: InMemoryMemoryStore is active. Data will not be persisted.") + + async def get_conversation(self, conversation_id: str, owner_id: str) -> Optional[EmmaConversation]: + convo = self._conversations.get(conversation_id) + # Enforce ownership check + if convo and convo.owner_id == owner_id: + return convo + return None + + async def save_conversation(self, conversation: EmmaConversation) -> None: + # Enforce ownership + if conversation.owner_id: + self._conversations[conversation.conversation_id] = conversation + + async def store_memory(self, record: MemoryRecord) -> None: + # Enforce ownership + if record.owner_id: + self._memories.append(record) + + async def recall_memory(self, embedding: List[float], owner_id: str, scope: List[str]) -> List[MemoryRecord]: + # Simple text match for testing, no real embedding search. + # Also filters on ownership. + recalled = [mem for mem in self._memories if mem.owner_id == owner_id] + return recalled[:5] # Return top 5 for simplicity diff --git a/opax-mcp/policy/__init__.py b/opax-mcp/policy/__init__.py new file mode 100644 index 0000000..44b2eb5 --- /dev/null +++ b/opax-mcp/policy/__init__.py @@ -0,0 +1 @@ +# This file makes the 'policy' directory a Python package. diff --git a/opax-mcp/policy/caller_context.py b/opax-mcp/policy/caller_context.py new file mode 100644 index 0000000..af19168 --- /dev/null +++ b/opax-mcp/policy/caller_context.py @@ -0,0 +1,66 @@ +""" +Derives a trusted CallerContext from a request. +""" +from typing import Dict, Any +from contracts.common import CallerContext, ToolRiskLevel + +# This is a placeholder for a future, more robust caller registry (e.g., Firestore). +# The key would be the hash of the API key or the IAP-verified user email. +_CALLER_REGISTRY: Dict[str, Dict[str, Any]] = { + "agent:perplexity": { + "profile": "operator", + "owner_id": "sa:perplexity", + "workspace_id": "ws:vauco", + "allowed_tool_policy": { + # This profile can read anything but requires approval for all writes + "default": ToolRiskLevel.REQUIRES_APPROVAL, + "read_only": ToolRiskLevel.READ_ONLY, + "propose_only": ToolRiskLevel.PROPOSE_ONLY, + "forbidden": ToolRiskLevel.FORBIDDEN, + } + }, + "user:chris.christiansen@vauco.no": { + "profile": "admin", + "owner_id": "user:chris.c", + "workspace_id": "ws:vauco", + "allowed_tool_policy": { + # Admin can do safe writes directly, but needs high approval for destructive actions + "default": ToolRiskLevel.REQUIRES_HIGH_APPROVAL, + "read_only": ToolRiskLevel.READ_ONLY, + "requires_approval": ToolRiskLevel.REQUIRES_APPROVAL, # Can do normal writes + } + } +} + +_DEFAULT_CALLER_CONTEXT = CallerContext( + caller_id="anonymous:unknown", + caller_type="system", + profile="readonly", + owner_id="system:public", + workspace_id="ws:public", + allowed_tool_policy={"default": ToolRiskLevel.READ_ONLY} +) + +def derive_caller_context(auth_identifier: str) -> CallerContext: + """ + Derives a CallerContext from a trusted, server-verified identifier. + + In a real implementation, `auth_identifier` would be the result of + authenticating a request (e.g., looking up an API key hash or using an + IAP-provided email address). + + Args: + auth_identifier: The trusted identifier for the caller. + + Returns: + A CallerContext object with the appropriate policies. + """ + caller_data = _CALLER_REGISTRY.get(auth_identifier) + if not caller_data: + return _DEFAULT_CALLER_CONTEXT + + return CallerContext( + caller_id=auth_identifier, + caller_type=auth_identifier.split(":")[0], # cheap trick + **caller_data + ) diff --git a/opax-mcp/policy/tool_policy.py b/opax-mcp/policy/tool_policy.py new file mode 100644 index 0000000..47f17b1 --- /dev/null +++ b/opax-mcp/policy/tool_policy.py @@ -0,0 +1,90 @@ +""" +Defines the Tool Policy Engine. +""" +from typing import Dict +from contracts.common import CallerContext, ToolRiskLevel + +# This matrix defines the risk level for each known tool. +# Tools not in this list will fall back to a default policy. +_TOOL_POLICY_MATRIX: Dict[str, ToolRiskLevel] = { + # Read Only + "get_health": "read_only", + "get_build_status": "read_only", + "get_state": "read_only", + "get_telemetry": "read_only", + "list_commits": "read_only", + "get_file": "read_only", + "list_open_issues": "read_only", + "list_emma_models": "read_only", + "run_emma": "read_only", + "call_emma": "read_only", + + # Requires Approval + "create_issue": "requires_approval", + "push_file": "requires_approval", + "create_branch": "requires_approval", + "create_commit": "requires_approval", + "create_pull_request": "requires_approval", + "trigger_build": "requires_approval", + "send_email": "requires_approval", + "set_billing_budget": "requires_approval", + + # Requires High Approval + "deploy_revision": "requires_high_approval", + "merge_pull_request": "requires_high_approval", + "IAM-policy change": "requires_high_approval", # Placeholder name + "firewall change": "requires_high_approval", # Placeholder name + "secret rotation": "requires_high_approval", # Placeholder name + "production database migration": "requires_high_approval", # Placeholder + "destructive repository action": "requires_high_approval", # Placeholder + + # Propose Only + "run_terminal": "propose_only", + "arbitrary gcloud": "propose_only", # Placeholder name + "arbitrary curl": "propose_only", # Placeholder name + "arbitrary SQL": "propose_only", # Placeholder name + + # Forbidden + "read_secret_value": "forbidden", + "token export": "forbidden", + "disable security controls": "forbidden", + "project deletion": "forbidden", + "VPC Service Controls disablement": "forbidden", +} + +class ToolPolicyEngine: + """A simple engine to check if a caller can execute a tool.""" + + def get_access_decision(self, caller: CallerContext, tool_name: str) -> str: + """ + Determines if a tool call is allowed, requires approval, or is denied. + + Returns: + One of: "allowed_directly", "requires_approval", "requires_high_approval", + "propose_only", "denied". + """ + tool_risk = _TOOL_POLICY_MATRIX.get(tool_name) + if not tool_risk or tool_risk == "forbidden": + return "denied" + + caller_permission = caller.allowed_tool_policy.get(tool_risk, caller.allowed_tool_policy.get("default")) + + if not caller_permission: + return "denied" + + # This logic determines the final decision based on the tool's risk + # and the caller's permission for that risk level. + if tool_risk == "read_only" and caller_permission == "read_only": + return "allowed_directly" + + if tool_risk == "requires_approval" and caller_permission in ["requires_approval", "requires_high_approval"]: + return "requires_approval" + + if tool_risk == "requires_high_approval" and caller_permission == "requires_high_approval": + return "requires_high_approval" + + if tool_risk == "propose_only" and caller_permission == "propose_only": + return "propose_only" + + return "denied" + diff --git a/opax-mcp/test_phase2a_foundation.py b/opax-mcp/test_phase2a_foundation.py new file mode 100644 index 0000000..bdacd60 --- /dev/null +++ b/opax-mcp/test_phase2a_foundation.py @@ -0,0 +1,119 @@ +import unittest +from unittest.mock import MagicMock, AsyncMock +from datetime import datetime, timezone +from pathlib import Path +import sys + +REPO_ROOT = Path(__file__).resolve().parents[1] +# Add the opax-mcp directory to the path to simulate the Docker container's layout +sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) + +from contracts.common import ( + EmmaConversation, MemoryRecord, ProposedToolAction, ApprovalRecord, ToolRiskLevel, MemoryType +) +from policy.caller_context import derive_caller_context +from policy.tool_policy import ToolPolicyEngine +from persistence.memory_store import InMemoryMemoryStore + + +class TestPhase2AContracts(unittest.TestCase): + + def test_models_have_schema_version(self): + """(C.4) Test that contracts have the schema_version field.""" + convo = EmmaConversation(owner_id="o", workspace_id="w", created_by="c") + self.assertEqual(convo.schema_version, "1.0.0") + + action = ProposedToolAction( + correlation_id="c1", tool_name="t1", parameters={}, + human_readable_summary="s1", risk_class="read_only", requires_approval=False + ) + self.assertEqual(action.schema_version, "1.0.0") + + def test_no_mutable_defaults_in_lists(self): + """Verifies that list fields are unique to each model instance.""" + c1 = EmmaConversation(owner_id="o", workspace_id="w", created_by="c") + c2 = EmmaConversation(owner_id="o", workspace_id="w", created_by="c") + c1.message_references.append("test") + self.assertNotEqual(c1.message_references, c2.message_references) + self.assertEqual(c2.message_references, []) + + def test_datetimes_are_timezone_aware(self): + """Verifies that default datetimes are timezone-aware.""" + c = EmmaConversation(owner_id="o", workspace_id="w", created_by="c") + self.assertIsNotNone(c.created_at.tzinfo) + + +class TestPhase2APolicy(unittest.TestCase): + + def setUp(self): + self.policy_engine = ToolPolicyEngine() + + def test_caller_context_derivation_and_least_privilege(self): + """Tests that known callers get correct profiles and unknown callers get a safe, read-only default.""" + op_context = derive_caller_context("agent:perplexity") + self.assertEqual(op_context.profile, "operator") + + unknown_context = derive_caller_context("some-random-key") + self.assertEqual(unknown_context.profile, "readonly") + self.assertEqual(unknown_context.allowed_tool_policy['default'], "read_only") + + def test_tool_policy_engine_enforces_all_levels(self): + """Tests that the policy engine correctly maps decisions for different user profiles and tool risks.""" + admin = derive_caller_context("user:chris.christiansen@vauco.no") + operator = derive_caller_context("agent:perplexity") + readonly = derive_caller_context("unknown:caller") + + # Test a read_only tool + self.assertEqual(self.policy_engine.get_access_decision(admin, "get_file"), "allowed_directly") + self.assertEqual(self.policy_engine.get_access_decision(operator, "get_file"), "allowed_directly") + self.assertEqual(self.policy_engine.get_access_decision(readonly, "get_file"), "allowed_directly") + + # Test a requires_approval tool + self.assertEqual(self.policy_engine.get_access_decision(admin, "create_issue"), "requires_approval") + self.assertEqual(self.policy_engine.get_access_decision(operator, "create_issue"), "requires_approval") + self.assertEqual(self.policy_engine.get_access_decision(readonly, "create_issue"), "denied") + + # Test a requires_high_approval tool + self.assertEqual(self.policy_engine.get_access_decision(admin, "deploy_revision"), "requires_high_approval") + self.assertEqual(self.policy_engine.get_access_decision(operator, "deploy_revision"), "denied") + + # Test a propose_only tool + self.assertEqual(self.policy_engine.get_access_decision(operator, "run_terminal"), "propose_only") + self.assertEqual(self.policy_engine.get_access_decision(admin, "run_terminal"), "denied") # Admins are not configured for this + + # Test a forbidden tool + self.assertEqual(self.policy_engine.get_access_decision(admin, "read_secret_value"), "denied") + + +class TestPhase2AMemoryStore(unittest.IsolatedAsyncioTestCase): + + def setUp(self): + self.store = InMemoryMemoryStore() + self.action = ProposedToolAction( + correlation_id="c1", tool_name="t1", parameters={}, + human_readable_summary="s1", risk_class="read_only", requires_approval=False + ) + + async def test_in_memory_store_enforces_owner_id(self): + """Tests that data is strictly partitioned by owner_id.""" + convo1 = EmmaConversation(owner_id="owner1", workspace_id="ws1", created_by="owner1") + await self.store.save_conversation(convo1) + + # Owner 1 can retrieve + self.assertIsNotNone(await self.store.get_conversation("c1", "owner1")) + # Owner 2 cannot retrieve + self.assertIsNone(await self.store.get_conversation("c1", "owner2")) + + async def test_approval_record_instantiation(self): + """(C.2) Tests that ApprovalRecord can be created with a real ProposedToolAction.""" + record = ApprovalRecord( + action=self.action, + expires_at=datetime.now(timezone.utc), + requested_by_caller_id="caller1", + workspace_id="ws1" + ) + self.assertEqual(record.action.tool_name, "t1") + + +if __name__ == '__main__': + unittest.main()