Spaces:
Runtime error
Runtime error
File size: 2,987 Bytes
595d60e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | 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)
|