91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""
|
|
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"
|
|
|