Spaces:
Runtime error
Runtime error
| from ..schemas.agent_spec import AgentSpec, ConnectionType | |
| from ..schemas.conversation import ValidationError, ValidationResult | |
| class ConnectionValidator: | |
| def validate(self, spec: AgentSpec) -> ValidationResult: | |
| errors = [] | |
| conn = spec.connection | |
| # 1. Validate required config keys based on type | |
| if conn.type == ConnectionType.API: | |
| if not conn.config or not conn.config.get("base_url"): | |
| errors.append(ValidationError( | |
| code="MISSING_CONNECTION_CONFIG", | |
| message="API connection requires 'base_url'", | |
| path="connection.config.base_url" | |
| )) | |
| elif conn.type == ConnectionType.DATABASE: | |
| if not conn.config or (not conn.config.get("connection_string") and not conn.config.get("host")): | |
| errors.append(ValidationError( | |
| code="MISSING_CONNECTION_CONFIG", | |
| message="Database connection requires 'connection_string' or 'host'", | |
| path="connection.config" | |
| )) | |
| elif conn.type == ConnectionType.WEBHOOK: | |
| if not conn.config or not conn.config.get("url"): | |
| errors.append(ValidationError( | |
| code="MISSING_CONNECTION_CONFIG", | |
| message="Webhook connection requires 'url'", | |
| path="connection.config.url" | |
| )) | |
| elif conn.type == ConnectionType.MCP: | |
| if not conn.config or not conn.config.get("command"): | |
| errors.append(ValidationError( | |
| code="MISSING_CONNECTION_CONFIG", | |
| message="MCP connection requires 'command'", | |
| path="connection.config.command" | |
| )) | |
| # 2. Check for exposed secrets/credentials | |
| if conn.config: | |
| credential_indicators = {"key", "secret", "password", "token", "credential", "cert", "pass", "pwd"} | |
| def scan_dict_for_secrets(d: dict, prefix: str): | |
| for k, v in d.items(): | |
| current_path = f"{prefix}.{k}" | |
| if isinstance(v, dict): | |
| scan_dict_for_secrets(v, current_path) | |
| elif isinstance(v, str): | |
| is_cred_key = any(ind in k.lower() for ind in credential_indicators) | |
| if is_cred_key and v.strip() and not v.startswith("$"): | |
| errors.append(ValidationError( | |
| code="EXPOSED_CREDENTIALS", | |
| message=f"Plaintext credential detected at '{current_path}'. Use a placeholder variable starting with '$' (e.g. '$API_KEY').", | |
| path=current_path | |
| )) | |
| scan_dict_for_secrets(conn.config, "connection.config") | |
| return ValidationResult(valid=len(errors) == 0, errors=errors) | |