"""Customer resource selection and private Sheets plans; no dispatch or platform credentials.""" import base64 import binascii import json import re from dataclasses import dataclass, field from typing import Any from uuid import UUID from pydantic import ValidationError from src.domain.connectors.catalog import get_action, slugs_implementing from src.domain.connectors.resource_contracts import RESOURCE_INPUTS @dataclass class ResourceConnection: connection_id: UUID connector_key: str config: dict = field(default_factory=dict) status: str = "active" has_secrets: bool = False scopes: set[str] = field(default_factory=set) permissions: dict[str, str] = field(default_factory=dict) execution_domain: str = "C" @dataclass class ResourcePlan: connection_id: UUID provider: str action: str input: dict[str, Any] class ResourceError(Exception): def __init__(self, code: str): self.code = code super().__init__(code) def validate_payload(capability: str, payload: dict) -> dict: model = RESOURCE_INPUTS.get(capability) if model is None: raise ResourceError("capability_not_invocable") try: return model.model_validate(payload).model_dump(exclude_none=True) except ValidationError as exc: raise ResourceError("invalid_input") from exc def _column(number: int) -> str: result = "" while number: number, remainder = divmod(number - 1, 26) result = chr(65 + remainder) + result return result def map_resource_input(capability: str, data: dict, binding: dict) -> dict: if capability.startswith("storage.file."): bucket = binding.get("bucket_name") if not isinstance(bucket, str) or not re.fullmatch( r"[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]", bucket ): raise ResourceError("resource_unbound") prefix = binding.get("prefix", "") key = data.get("key", data.get("prefix", "")) for part in (prefix, key): if ( not isinstance(part, str) or part.startswith("/") or re.search(r"[\\\x00-\x1f\x7f]", part) or any(segment in {".", ".."} for segment in part.split("/")) ): raise ResourceError("invalid_input") addressed_key = (prefix.rstrip("/") + "/" if prefix else "") + key if len(addressed_key) > 1024: raise ResourceError("invalid_input") result = {"bucket_name": bucket} if capability == "storage.file.list": result.update(prefix=addressed_key, max_keys=data["limit"]) if data.get("cursor"): result["continuation_token"] = data["cursor"] else: result["key"] = addressed_key if capability == "storage.file.put": try: decoded = base64.b64decode(data["content_base64"], validate=True) if ( len(decoded) > 262144 or base64.b64encode(decoded).decode() != data["content_base64"] or "\r" in data["content_type"] or "\n" in data["content_type"] ): raise ValueError("invalid content") except (ValueError, binascii.Error) as exc: raise ResourceError("invalid_input") from exc result.update(content_base64=data["content_base64"], content_type=data["content_type"]) return result resource_id, sheet = binding.get("spreadsheet_id"), binding.get("sheet_name") start, columns = binding.get("start_row", 1), binding.get("column_count") if ( not isinstance(resource_id, str) or not resource_id or not isinstance(sheet, str) or not sheet or type(start) is not int or not 1 <= start <= 1_000_000 or type(columns) is not int or not 1 <= columns <= 100 ): raise ResourceError("resource_unbound") rows = data.get("rows") if rows and any(len(row) != columns for row in rows): raise ResourceError("invalid_input") if rows and len(json.dumps(rows).encode()) > 500000: raise ResourceError("invalid_input") quoted = "'" + sheet.replace("'", "''") + "'" first = start + data.get("offset", 0) last = first + (len(rows) if rows else data["limit"]) - 1 cell_range = f"A{first}:{_column(columns)}{last}" if capability.endswith(".append"): cell_range = f"A{start}:{_column(columns)}" result = {"spreadsheet_id": resource_id, "range": f"{quoted}!{cell_range}"} if rows: result.update(values=rows, value_input_option="RAW") return result def resolve_resource( capability: str, payload: dict, connections: list[ResourceConnection], *, enabled_capabilities: set[str], provider_hint: str | None = None, binding_connection_id: UUID | None = None, ) -> ResourcePlan: data = validate_payload(capability, payload) resource_kind = "storage" if capability.startswith("storage.file.") else "spreadsheet" alias = data[resource_kind] implementations = slugs_implementing(capability) candidates = [] rejected = {} for row in connections: if row.execution_domain != "C": continue if provider_hint and row.connector_key != provider_hint: continue actions = [slug for slug in implementations if get_action(slug)[0].key == row.connector_key] if not actions: continue if len(actions) != 1: raise ResourceError("binding_ambiguous") action = actions[0] definition = get_action(action)[1] reason = None if row.status != "active": reason = "connection_inactive" elif not row.has_secrets: reason = "credentials_missing" elif not set(definition.required_scopes) <= row.scopes: reason = "scope_missing" elif not set(definition.required_capabilities) <= enabled_capabilities: reason = "capability_not_enabled" elif row.permissions.get(action, "deny") == "deny": reason = "connector_action_denied" if reason: rejected[row.connection_id] = reason continue mapping = row.config.get(f"{resource_kind}_map") binding = mapping.get(alias) if isinstance(mapping, dict) else None candidates.append((row, action, binding)) if binding_connection_id: if binding_connection_id in rejected: raise ResourceError(rejected[binding_connection_id]) candidates = [item for item in candidates if item[0].connection_id == binding_connection_id] if not candidates: raise ResourceError("capability_not_bound") matching = [item for item in candidates if isinstance(item[2], dict)] if len(matching) > 1: raise ResourceError("binding_ambiguous") if not matching: if len(candidates) > 1: raise ResourceError("binding_ambiguous") raise ResourceError("resource_unbound" if candidates else "capability_not_bound") row, action, binding = matching[0] return ResourcePlan( row.connection_id, row.connector_key, action, map_resource_input(capability, data, binding) )