Security Analysis - Crystal Agents

def approval_resolver(agent, tool, context, kwargs) -> bool:
    tool_config = policy.policy.get("tools", {}).get(tool.name, {})
    if not tool_config.get("requires_approval", False):
        return True
    return False  # Always denies approval-required tools

def approval_resolver(agent, tool, context, kwargs) -> bool:
    tool_config = policy.policy.get("tools", {}).get(tool.name, {})
    if not tool_config.get("requires_approval", False):
        return True
    # Check if already approved in this session
    action_key = f"{context.run_id}:{agent.name}:{tool.name}"
    return context.metadata.get("pre_approved", False)  # or check approval store

# Option 1: Basic HTTP Auth
class AuthenticatedDashboardHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        auth = self.headers.get("Authorization", "")
        if not self._verify_auth(auth):
            self.send_response(401)
            self.send_header("WWW-Authenticate", 'Basic realm="Crystal Dashboard"')
            self.end_headers()
            return

# Option 2: Session-based auth with login page
# Option 3: API key in headers

if chat_id != str(telegram.chat_id):
    return False

# Verify user is in allowed list
allowed_users = telegram.allowed_users or []
if chat_id not in [str(id) for id in allowed_users]:
    return False

# Or use bot commands only from admin
admin_user_id = telegram.admin_user_id
if message.get("from", {}).get("id") != admin_user_id:
    self._runtime.notifier.send_message("Unauthorized")
    return False

def _resolve_target(self, target: str) -> Path:
    # Only checks if resolved path is under workspace
    resolved = (Path.cwd() / candidate).resolve()
    workspace_root = Path.cwd().resolve()
    if workspace_root not in resolved.parents and resolved != workspace_root:
        raise ValueError("RepoWriterTool target must stay within the workspace")
    return resolved

def _resolve_target(self, target: str) -> Path:
    candidate = Path(target)
    if candidate.is_absolute():
        raise ValueError("RepoWriterTool requires a relative target path")
    
    resolved = (Path.cwd() / candidate).resolve()
    
    # Follow symlinks and check again
    try:
        real_path = resolved.resolve()
    except (OSError, RuntimeError):
        raise ValueError(f"Cannot resolve path: {target}")
    
    workspace_root = Path.cwd().resolve()
    if workspace_root not in real_path.parents and real_path != workspace_root:
        raise ValueError("RepoWriterTool target must stay within the workspace")
    return real_path

class AgentPromptSanitizer:
    FORBIDDEN_PATTERNS = [
        r"ignore previous instructions",
        r"forget everything",
        r"system prompt",
    ]
    
    @classmethod
    def sanitize(cls, prompt: str) -> str:
        import re
        for pattern in cls.FORBIDDEN_PATTERNS:
            if re.search(pattern, prompt, re.IGNORECASE):
                raise ValueError("Potentially malicious prompt detected")
        return prompt

def sanitize_for_logging(metadata: dict) -> dict:
    sensitive_keys = ["password", "token", "api_key", "secret"]
    sanitized = {}
    for k, v in metadata.items():
        if any(s in k.lower() for s in sensitive_keys):
            sanitized[k] = "***REDACTED***"
        else:
            sanitized[k] = v
    return sanitized

class RateLimiter:
    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = {}
    
    def is_allowed(self, identifier: str) -> bool:
        now = time.time()
        self.requests[identifier] = [
            t for t in self.requests.get(identifier, [])
            if now - t < self.window
        ]
        if len(self.requests[identifier]) >= self.max_requests:
            return False
        self.requests[identifier].append(now)
        return True