tiphys.security

 44class PolicyEngine:
 45    """
 46    Multi-layer security policy engine.
 47
 48    A tool must pass ALL layers to be allowed (intersection logic).
 49    Layers are checked in order: global -> provider -> agent -> profile -> group -> sandbox.
 50    """
 51
 52    LAYER_ORDER = ["global", "provider", "agent", "profile", "group", "sandbox"]
 53
 54    def __init__(self, config: TiphysConfig | None = None) -> None:
 55        """Initialize policy engine with configuration."""
 56        self._config = config
 57        self._layers: dict[str, PolicyLayer] = {}
 58
 59    def add_layer(self, layer: PolicyLayer) -> None:
 60        """Add or update a policy layer."""
 61        self._layers[layer.name] = layer
 62
 63    def remove_layer(self, name: str) -> None:
 64        """Remove a policy layer."""
 65        self._layers.pop(name, None)
 66
 67    def get_layer(self, name: str) -> PolicyLayer | None:
 68        """Get a policy layer by name."""
 69        return self._layers.get(name)
 70
 71    def check(
 72        self,
 73        tool_name: str,
 74        context: ToolContext | None = None,
 75    ) -> ToolPermission:
 76        """
 77        Check if a tool is allowed in the given context.
 78
 79        Args:
 80            tool_name: Name of the tool to check.
 81            context: Optional context for policy evaluation.
 82
 83        Returns:
 84            ToolPermission indicating whether the tool is allowed, denied, or requires approval.
 85        """
 86        # Get all applicable layers in order
 87        layers = self._get_applicable_layers(context)
 88        return self._resolve(tool_name, layers)
 89
 90    def _resolve(
 91        self,
 92        tool_name: str,
 93        layers: list[PolicyLayer],
 94    ) -> ToolPermission:
 95        """Resolve effective permission across all layers."""
 96        needs_ask = False
 97
 98        for layer in layers:
 99            result = self._check_layer(tool_name, layer)
100
101            # Denied by any layer = denied overall
102            if result == ToolPermission.DENIED:
103                return ToolPermission.DENIED
104
105            # Track if any layer requires asking
106            if result == ToolPermission.ASK:
107                needs_ask = True
108
109        return ToolPermission.ASK if needs_ask else ToolPermission.ALLOWED
110
111    def _check_layer(
112        self,
113        tool_name: str,
114        layer: PolicyLayer,
115    ) -> ToolPermission:
116        """Check a single policy layer."""
117        # Deny takes priority
118        if self._matches(tool_name, layer.deny):
119            return ToolPermission.DENIED
120
121        # If allow list exists, must match
122        if layer.allow is not None and not self._matches(tool_name, layer.allow):
123            return ToolPermission.DENIED
124
125        # Check if tool requires approval
126        if self._matches(tool_name, layer.ask):
127            return ToolPermission.ASK
128
129        return ToolPermission.ALLOWED
130
131    def _matches(self, tool: str, patterns: list[str]) -> bool:
132        """Check if tool name matches any pattern."""
133        return any(fnmatch(tool, pattern) for pattern in patterns)
134
135    def _get_applicable_layers(self, context: ToolContext | None) -> list[PolicyLayer]:
136        """Get all applicable policy layers for the given context."""
137        layers: list[PolicyLayer] = []
138
139        for layer_name in self.LAYER_ORDER:
140            if layer_name in self._layers:
141                layer = self._layers[layer_name]
142
143                # Filter layers based on context
144                if context is not None:
145                    # Skip sandbox layer if not sandboxed
146                    if layer_name == "sandbox" and not context.is_sandboxed:
147                        continue
148
149                    # Skip guest profile layer for trusted users
150                    if layer_name == "profile" and context.profile != "guest":
151                        continue
152
153                layers.append(layer)
154
155        return layers

24class PolicyLayer(BaseModel):
25    """A single policy layer with allow/deny/ask rules."""
26
27    name: str
28    allow: list[str] | None = None  # None = allow all not denied
29    deny: list[str] = Field(default_factory=list)
30    ask: list[str] = Field(default_factory=list)

33class ToolContext(BaseModel):
34    """Context for policy evaluation."""
35
36    agent_id: str
37    channel: str = "webchat"
38    peer_id: str | None = None
39    is_sandboxed: bool = False
40    profile: str = "trusted"  # "trusted" or "guest"
41    group_id: str | None = None

16class ToolPermission(str, Enum):
17    """Result of a policy check."""
18
19    ALLOWED = "allowed"
20    DENIED = "denied"
21    ASK = "ask"  # Requires user approval

158def create_default_policy(config: TiphysConfig) -> PolicyEngine:
159    """
160    Create a policy engine with default layers based on configuration.
161
162    Args:
163        config: Tiphys configuration.
164
165    Returns:
166        Configured PolicyEngine instance.
167    """
168    engine = PolicyEngine(config)
169
170    # Global layer based on exec_mode
171    if config.exec_mode == "deny":
172        # Deny all command execution
173        global_layer = PolicyLayer(
174            name="global",
175            deny=["run_command", "shell", "exec"],
176        )
177    elif config.exec_mode == "allowlist":
178        # Only allow specific commands (handled at command level, not tool level)
179        global_layer = PolicyLayer(
180            name="global",
181            ask=["run_command"],  # Require approval for commands
182        )
183    else:  # full
184        # Allow everything
185        global_layer = PolicyLayer(
186            name="global",
187            allow=["*"],
188        )
189
190    engine.add_layer(global_layer)
191
192    # Guest profile layer - more restrictive
193    guest_layer = PolicyLayer(
194        name="profile",
195        deny=["run_command", "write_file", "delete_*"],
196        allow=["read_file", "list_directory", "search"],
197    )
198    engine.add_layer(guest_layer)
199
200    # Sandbox layer - when running in container
201    sandbox_layer = PolicyLayer(
202        name="sandbox",
203        deny=["network_*", "sudo_*"],
204    )
205    engine.add_layer(sandbox_layer)
206
207    return engine

145@dataclass
146class CommandAnalysis:
147    """Analysis of a shell command."""
148
149    raw_command: str
150    binaries: list[str] = field(default_factory=list)
151    has_pipes: bool = False
152    has_chaining: bool = False  # &&, ||, ;
153    has_redirects: bool = False
154    has_subshell: bool = False  # $() or ``
155    has_wildcards: bool = False
156    risk_level: RiskLevel = RiskLevel.LOW
157    reasons: list[str] = field(default_factory=list)
158    is_allowed: bool = True

14class RiskLevel(str, Enum):
15    """Risk level of a command."""
16
17    LOW = "low"
18    MEDIUM = "medium"
19    HIGH = "high"
20    CRITICAL = "critical"

161def analyze_command(cmd: str, allowed_commands: list[str] | None = None) -> CommandAnalysis:
162    """
163    Analyze a shell command for security risks.
164
165    Args:
166        cmd: The shell command to analyze.
167        allowed_commands: Optional list of allowed command binaries.
168
169    Returns:
170        CommandAnalysis with risk assessment.
171    """
172    analysis = CommandAnalysis(raw_command=cmd)
173    reasons: list[str] = []
174
175    # Check for pipes
176    analysis.has_pipes = "|" in cmd
177
178    # Check for command chaining
179    analysis.has_chaining = any(op in cmd for op in ["&&", "||", ";"])
180
181    # Check for redirects
182    analysis.has_redirects = any(op in cmd for op in [">", ">>", "<", "2>", "&>"])
183
184    # Check for subshells
185    analysis.has_subshell = "$(" in cmd or "`" in cmd
186
187    # Check for wildcards
188    analysis.has_wildcards = any(c in cmd for c in ["*", "?", "["])
189
190    # Parse command to extract binaries
191    binaries = _extract_binaries(cmd)
192    analysis.binaries = binaries
193
194    # Determine risk level
195    risk_level = RiskLevel.LOW
196
197    # Subshells are high risk (arbitrary code execution)
198    if analysis.has_subshell:
199        risk_level = _max_risk(risk_level, RiskLevel.HIGH)
200        reasons.append("Contains subshell execution")
201
202    # Pipes and chaining increase risk
203    if analysis.has_pipes or analysis.has_chaining:
204        risk_level = _max_risk(risk_level, RiskLevel.MEDIUM)
205        reasons.append("Contains pipes or command chaining")
206
207    # Redirects can overwrite files
208    if analysis.has_redirects:
209        risk_level = _max_risk(risk_level, RiskLevel.MEDIUM)
210        reasons.append("Contains output redirection")
211
212    # Check each binary
213    for binary in binaries:
214        binary_name = Path(binary).name
215
216        if binary_name in DANGEROUS_BINARIES:
217            risk_level = RiskLevel.CRITICAL
218            reasons.append(f"Uses dangerous binary: {binary_name}")
219        elif binary_name not in SAFE_BINARIES:
220            risk_level = _max_risk(risk_level, RiskLevel.MEDIUM)
221            reasons.append(f"Uses unknown binary: {binary_name}")
222
223    # Check against allowlist if provided
224    if allowed_commands is not None:
225        for binary in binaries:
226            binary_name = Path(binary).name
227            if binary_name not in allowed_commands:
228                analysis.is_allowed = False
229                reasons.append(f"Binary not in allowlist: {binary_name}")
230
231    analysis.risk_level = risk_level
232    analysis.reasons = reasons
233
234    return analysis

278def is_command_safe(cmd: str, allowed_commands: list[str] | None = None) -> bool:
279    """
280    Quick check if a command is considered safe.
281
282    Args:
283        cmd: The shell command to check.
284        allowed_commands: Optional list of allowed command binaries.
285
286    Returns:
287        True if the command is considered safe.
288    """
289    analysis = analyze_command(cmd, allowed_commands)
290    return analysis.risk_level in (RiskLevel.LOW, RiskLevel.MEDIUM) and analysis.is_allowed

 31class AskProtocol:
 32    """
 33    Interactive permission request system.
 34
 35    Allows the agent to request user approval for risky operations.
 36    Requests are sent via the event bus and responses are awaited with timeout.
 37    """
 38
 39    DEFAULT_TIMEOUT = 300.0  # 5 minutes
 40
 41    def __init__(self, event_bus: EventBus) -> None:
 42        """
 43        Initialize the ask protocol.
 44
 45        Args:
 46            event_bus: Event bus for sending requests and receiving responses.
 47        """
 48        self.event_bus = event_bus
 49        self._pending: dict[str, asyncio.Future[bool]] = {}
 50
 51        # Subscribe to permission responses
 52        self.event_bus.subscribe("permission_response", self._handle_response)
 53
 54    async def request_permission(
 55        self,
 56        action: str,
 57        details: str,
 58        agent_id: str,
 59        tool_name: str | None = None,
 60        arguments: dict[str, object] | None = None,
 61        timeout: float | None = None,
 62    ) -> bool:
 63        """
 64        Request user permission for an action.
 65
 66        Args:
 67            action: Type of action (e.g., "run_command", "write_file").
 68            details: Human-readable description of what will happen.
 69            agent_id: ID of the agent requesting permission.
 70            tool_name: Optional name of the tool being used.
 71            arguments: Optional tool arguments for context.
 72            timeout: Timeout in seconds. Defaults to 5 minutes.
 73
 74        Returns:
 75            True if approved, False if denied or timeout.
 76        """
 77        if timeout is None:
 78            timeout = self.DEFAULT_TIMEOUT
 79
 80        request = PendingRequest(
 81            action=action,
 82            details=details,
 83            agent_id=agent_id,
 84            tool_name=tool_name,
 85            arguments=arguments,
 86        )
 87
 88        # Create a future to wait for response
 89        future: asyncio.Future[bool] = asyncio.get_event_loop().create_future()
 90        self._pending[request.request_id] = future
 91
 92        try:
 93            # Emit permission request event
 94            event = PermissionRequestEvent(
 95                request_id=request.request_id,
 96                action=action,
 97                details=details,
 98                agent_id=agent_id,
 99            )
100            await self.event_bus.broadcast(event)
101
102            await logger.ainfo(
103                "Permission requested",
104                request_id=request.request_id,
105                action=action,
106                agent_id=agent_id,
107            )
108
109            # Wait for response with timeout
110            approved = await asyncio.wait_for(future, timeout=timeout)
111
112            await logger.ainfo(
113                "Permission response received",
114                request_id=request.request_id,
115                approved=approved,
116            )
117
118            return approved
119
120        except TimeoutError:
121            await logger.awarning(
122                "Permission request timed out",
123                request_id=request.request_id,
124                timeout=timeout,
125            )
126            return False
127
128        except asyncio.CancelledError:
129            await logger.awarning(
130                "Permission request cancelled",
131                request_id=request.request_id,
132            )
133            return False
134
135        finally:
136            # Clean up pending request
137            self._pending.pop(request.request_id, None)
138
139    def handle_response(self, request_id: str, approved: bool) -> bool:
140        """
141        Handle a permission response from the user.
142
143        This is called when the user approves or denies a request.
144
145        Args:
146            request_id: ID of the request being responded to.
147            approved: Whether the request was approved.
148
149        Returns:
150            True if the request was found and handled, False otherwise.
151        """
152        if request_id not in self._pending:
153            return False
154
155        future = self._pending[request_id]
156        if not future.done():
157            future.set_result(approved)
158        return True
159
160    async def _handle_response(self, event: "BaseEvent") -> None:
161        """Handle permission response events from the bus."""
162        # Type narrowing - we only subscribe to permission_response events
163        if isinstance(event, PermissionResponseEvent):
164            self.handle_response(event.request_id, event.approved)
165
166    def cancel_request(self, request_id: str) -> bool:
167        """
168        Cancel a pending permission request.
169
170        Args:
171            request_id: ID of the request to cancel.
172
173        Returns:
174            True if the request was found and cancelled, False otherwise.
175        """
176        if request_id not in self._pending:
177            return False
178
179        future = self._pending.pop(request_id)
180        if not future.done():
181            future.cancel()
182        return True
183
184    def cancel_all(self) -> int:
185        """
186        Cancel all pending permission requests.
187
188        Returns:
189            Number of requests cancelled.
190        """
191        count = 0
192        for request_id in list(self._pending.keys()):
193            if self.cancel_request(request_id):
194                count += 1
195        return count
196
197    @property
198    def pending_count(self) -> int:
199        """Number of pending permission requests."""
200        return len(self._pending)

20class PendingRequest(BaseModel):
21    """A pending permission request."""
22
23    request_id: str = Field(default_factory=lambda: str(uuid4()))
24    action: str
25    details: str
26    agent_id: str
27    tool_name: str | None = None
28    arguments: dict[str, object] | None = None

204async def ask_permission(
205    event_bus: EventBus,
206    action: str,
207    details: str,
208    agent_id: str,
209    timeout: float = 300.0,
210) -> bool:
211    """
212    Request user permission for an action (convenience function).
213
214    Args:
215        event_bus: Event bus for communication.
216        action: Type of action.
217        details: Human-readable description.
218        agent_id: ID of the requesting agent.
219        timeout: Timeout in seconds.
220
221    Returns:
222        True if approved, False if denied or timeout.
223    """
224    protocol = AskProtocol(event_bus)
225    return await protocol.request_permission(
226        action=action,
227        details=details,
228        agent_id=agent_id,
229        timeout=timeout,
230    )