67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
"""
|
|
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
|
|
)
|