Skills › Scientific Agent Skills › protocolsio-integration › _common.py
_common.py
614 lines · skills/protocolsio-integration/scripts/_common.py at 5f2bc03
1"""Shared safety primitives for the protocols.io helper scripts.
3The helpers use only the Python standard library. They never load ``.env``
4files, never accept credentials as command-line arguments, and never follow
5HTTP redirects.
6"""
8from __future__ import annotations
10import json
11import math
12import os
13import re
14import stat
15import time
16import urllib.error
17import urllib.parse
18import urllib.request
19from dataclasses import dataclass
20from pathlib import Path
21from typing import Any, Callable, Mapping, MutableMapping, Sequence
24DEFAULT_ORIGIN = "https://www.protocols.io"
25ACCESS_TOKEN_ENV = "PROTOCOLS_IO_ACCESS_TOKEN"
26CREDENTIAL_ENV_NAMES = (ACCESS_TOKEN_ENV,)
28MAX_LOCAL_JSON_BYTES = 2_000_000
29MAX_JSON_RESPONSE_BYTES = 4_000_000
30MAX_PDF_RESPONSE_BYTES = 25_000_000
31MAX_ERROR_BYTES = 64_000
32MAX_RETRIES = 2
33MAX_RETRY_AFTER_SECONDS = 30.0
34MIN_TIMEOUT_SECONDS = 1.0
35MAX_TIMEOUT_SECONDS = 60.0
36DEFAULT_TIMEOUT_SECONDS = 15.0
37MAX_REMOTE_STRING_CHARS = 4_000
38MAX_REMOTE_COLLECTION_ITEMS = 500
39MAX_REMOTE_DEPTH = 12
41_CORE_HOSTS = frozenset({"protocols.io", "www.protocols.io"})
42_TENANT_HOST_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.protocols\.io$")
43_PROTOCOL_ID_RE = re.compile(
44 r"^(?:10\.17504/protocols\.io\.[A-Za-z0-9][A-Za-z0-9._-]{0,240}|"
45 r"[1-9][0-9]*|[A-Za-z0-9][A-Za-z0-9._-]{0,254})"
46 r"(?:/(?:v[1-9][0-9]*|latest))?$"
47)
48_GUID_RE = re.compile(r"^[A-Fa-f0-9]{32}$")
49_INTEGER_RE = re.compile(r"^(?:0|[1-9][0-9]*)$")
50_CONTROL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
51_SENSITIVE_KEY_PARTS = (
52 "authorization",
53 "credential",
54 "password",
55 "secret",
56 "signature",
57 "access_token",
58 "refresh_token",
59 "awsaccesskey",
60 "policy",
61)
64class SafetyError(ValueError):
65 """A local, user-correctable safety or validation failure."""
68class ApiError(RuntimeError):
69 """A bounded API failure that deliberately excludes credentials and bodies."""
71 def __init__(
72 self,
73 message: str,
74 *,
75 http_status: int | None = None,
76 api_status: int | str | None = None,
77 ) -> None:
78 super().__init__(message)
79 self.http_status = http_status
80 self.api_status = api_status
83@dataclass(frozen=True)
84class HttpResult:
85 """A bounded HTTP response."""
87 status: int
88 headers: Mapping[str, str]
89 body: bytes
90 url: str
91 attempts: int
94class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
95 """Reject every redirect so credentials cannot cross origins."""
97 def redirect_request( # type: ignore[override]
98 self,
99 req: urllib.request.Request,
100 fp: Any,
101 code: int,
102 msg: str,
103 headers: Mapping[str, str],
104 newurl: str,
105 ) -> None:
106 del req, fp, code, msg, headers, newurl
107 return None
110def _reject_constant(value: str) -> None:
111 raise SafetyError(f"non-finite JSON number is forbidden: {value}")
114def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
115 result: dict[str, Any] = {}
116 for key, value in pairs:
117 if key in result:
118 raise SafetyError(f"duplicate JSON key: {key}")
119 result[key] = value
120 return result
123def parse_json_bytes(raw: bytes, *, source: str = "response") -> Any:
124 """Parse strict UTF-8 JSON with duplicate-key and NaN rejection."""
126 try:
127 text = raw.decode("utf-8")
128 except UnicodeDecodeError as exc:
129 raise SafetyError(f"{source} is not UTF-8 JSON") from exc
130 try:
131 return json.loads(
132 text,
133 object_pairs_hook=_unique_object,
134 parse_constant=_reject_constant,
135 )
136 except json.JSONDecodeError as exc:
137 raise SafetyError(
138 f"invalid JSON in {source} at line {exc.lineno}, column {exc.colno}"
139 ) from exc
142def _path_inside_cwd(raw_path: str, *, must_exist: bool) -> Path:
143 base = Path.cwd().resolve()
144 candidate = Path(raw_path)
145 unresolved = candidate if candidate.is_absolute() else base / candidate
146 try:
147 relative = unresolved.relative_to(base)
148 except ValueError as exc:
149 raise SafetyError(
150 "path must stay inside the current working directory"
151 ) from exc
152 if ".." in relative.parts:
153 raise SafetyError("parent-directory traversal is forbidden")
155 cursor = base
156 parts = relative.parts[:-1] if not must_exist else relative.parts
157 for part in parts:
158 cursor /= part
159 if cursor.is_symlink():
160 raise SafetyError(f"symlink paths are forbidden: {raw_path}")
162 resolved = unresolved.resolve(strict=False)
163 try:
164 resolved.relative_to(base)
165 except ValueError as exc:
166 raise SafetyError("resolved path escapes the working directory") from exc
167 return resolved
170def safe_input_path(
171 raw_path: str,
172 *,
173 suffixes: Sequence[str],
174 max_bytes: int,
175) -> Path:
176 """Return a bounded, regular, non-symlink input path under the CWD."""
178 path = _path_inside_cwd(raw_path, must_exist=True)
179 if path.is_symlink() or not path.exists() or not path.is_file():
180 raise SafetyError(f"input is not a regular non-symlink file: {raw_path}")
181 if suffixes and path.suffix.lower() not in {suffix.lower() for suffix in suffixes}:
182 raise SafetyError("input suffix must be one of: " + ", ".join(sorted(suffixes)))
183 size = path.stat().st_size
184 if size > max_bytes:
185 raise SafetyError(f"input exceeds the local safety cap of {max_bytes} bytes")
186 return path
189def load_local_json(
190 raw_path: str,
191 *,
192 max_bytes: int = MAX_LOCAL_JSON_BYTES,
193) -> Any:
194 path = safe_input_path(raw_path, suffixes=(".json",), max_bytes=max_bytes)
195 return parse_json_bytes(path.read_bytes(), source=str(path))
198def safe_output_path(raw_path: str, *, suffix: str) -> Path:
199 """Validate a new output path below the CWD without creating it."""
201 path = _path_inside_cwd(raw_path, must_exist=False)
202 if path.suffix.lower() != suffix.lower():
203 raise SafetyError(f"output must have a {suffix} suffix")
204 parent = path.parent
205 if not parent.exists() or not parent.is_dir() or parent.is_symlink():
206 raise SafetyError("output parent must be an existing non-symlink directory")
207 if path.exists() or path.is_symlink():
208 raise SafetyError("refusing to overwrite an existing output path")
209 return path
212def write_private_bytes(path: Path, data: bytes) -> None:
213 """Create a private output file atomically with collision protection."""
215 descriptor = os.open(
216 path,
217 os.O_WRONLY | os.O_CREAT | os.O_EXCL,
218 stat.S_IRUSR | stat.S_IWUSR,
219 )
220 try:
221 with os.fdopen(descriptor, "wb") as handle:
222 handle.write(data)
223 except Exception:
224 try:
225 path.unlink()
226 except OSError:
227 pass
228 raise
231def clean_text(value: str, *, max_chars: int = MAX_REMOTE_STRING_CHARS) -> str:
232 """Remove control characters and bound untrusted text."""
234 cleaned = _CONTROL_RE.sub("\ufffd", value)
235 if len(cleaned) <= max_chars:
236 return cleaned
237 return cleaned[:max_chars] + f"\u2026[truncated {len(cleaned) - max_chars} chars]"
240def _sensitive_key(key: str) -> bool:
241 normalized = key.lower().replace("-", "_")
242 return any(part in normalized for part in _SENSITIVE_KEY_PARTS)
245def sanitize_untrusted(
246 value: Any,
247 *,
248 depth: int = 0,
249 max_string_chars: int = MAX_REMOTE_STRING_CHARS,
250 max_items: int = MAX_REMOTE_COLLECTION_ITEMS,
251) -> Any:
252 """Redact secrets and bound arbitrary remote or user-provided structures."""
254 if depth > MAX_REMOTE_DEPTH:
255 return {"truncated": "maximum nesting depth reached"}
256 if value is None or isinstance(value, (bool, int)):
257 return value
258 if isinstance(value, float):
259 return value if math.isfinite(value) else "[NON_FINITE_REDACTED]"
260 if isinstance(value, str):
261 return clean_text(value, max_chars=max_string_chars)
262 if isinstance(value, Mapping):
263 result: MutableMapping[str, Any] = {}
264 entries = list(value.items())
265 for key, item in entries[:max_items]:
266 text_key = clean_text(str(key), max_chars=256)
267 if _sensitive_key(text_key):
268 result[text_key] = "[REDACTED]"
269 else:
270 result[text_key] = sanitize_untrusted(
271 item,
272 depth=depth + 1,
273 max_string_chars=max_string_chars,
274 max_items=max_items,
275 )
276 if len(entries) > max_items:
277 result["_truncated_items"] = len(entries) - max_items
278 return dict(result)
279 if isinstance(value, (list, tuple)):
280 result = [
281 sanitize_untrusted(
282 item,
283 depth=depth + 1,
284 max_string_chars=max_string_chars,
285 max_items=max_items,
286 )
287 for item in value[:max_items]
288 ]
289 if len(value) > max_items:
290 result.append({"_truncated_items": len(value) - max_items})
291 return result
292 return clean_text(repr(value), max_chars=max_string_chars)
295def emit_json(payload: Any, *, stream: Any = None) -> None:
296 if stream is None:
297 import sys
299 stream = sys.stdout
300 print(
301 json.dumps(payload, indent=2, sort_keys=True, allow_nan=False),
302 file=stream,
303 )
306def emit_error(error: Exception) -> int:
307 import sys
309 payload: dict[str, Any] = {
310 "ok": False,
311 "error": type(error).__name__,
312 "message": clean_text(str(error), max_chars=500),
313 }
314 if isinstance(error, ApiError):
315 payload["http_status"] = error.http_status
316 payload["api_status"] = error.api_status
317 emit_json(payload, stream=sys.stderr)
318 return 2
321def validate_timeout(value: float) -> float:
322 if (
323 not math.isfinite(value)
324 or not MIN_TIMEOUT_SECONDS <= value <= MAX_TIMEOUT_SECONDS
325 ):
326 raise SafetyError(
327 f"timeout must be between {MIN_TIMEOUT_SECONDS:g} and "
328 f"{MAX_TIMEOUT_SECONDS:g} seconds"
329 )
330 return value
333def validate_retries(value: int) -> int:
334 if isinstance(value, bool) or not 0 <= value <= MAX_RETRIES:
335 raise SafetyError(f"retries must be between 0 and {MAX_RETRIES}")
336 return value
339def is_official_host(host: str, *, allow_tenant: bool = False) -> bool:
340 normalized = host.rstrip(".").lower()
341 if normalized in _CORE_HOSTS:
342 return True
343 return allow_tenant and _TENANT_HOST_RE.fullmatch(normalized) is not None
346def validate_origin(raw_origin: str, *, allow_tenant: bool = False) -> str:
347 parsed = urllib.parse.urlsplit(raw_origin)
348 try:
349 port = parsed.port
350 except ValueError as exc:
351 raise SafetyError("origin contains an invalid port") from exc
352 if (
353 parsed.scheme != "https"
354 or not parsed.hostname
355 or parsed.username is not None
356 or parsed.password is not None
357 or port not in (None, 443)
358 or parsed.query
359 or parsed.fragment
360 or parsed.path not in ("", "/")
361 ):
362 raise SafetyError(
363 "origin must be an HTTPS protocols.io origin without credentials, "
364 "path, query, fragment, or non-default port"
365 )
366 if not is_official_host(parsed.hostname, allow_tenant=allow_tenant):
367 raise SafetyError("origin host is not on the official protocols.io allowlist")
368 return f"https://{parsed.hostname.lower()}"
371def validate_remote_url(
372 raw_url: str,
373 *,
374 allow_tenant: bool = False,
375 allowed_paths: Sequence[str] = ("/api/", "/view/"),
376) -> str:
377 parsed = urllib.parse.urlsplit(raw_url)
378 try:
379 port = parsed.port
380 except ValueError as exc:
381 raise SafetyError("URL contains an invalid port") from exc
382 if (
383 parsed.scheme != "https"
384 or not parsed.hostname
385 or parsed.username is not None
386 or parsed.password is not None
387 or port not in (None, 443)
388 or parsed.fragment
389 ):
390 raise SafetyError("remote URL must be credential-free HTTPS on port 443")
391 if not is_official_host(parsed.hostname, allow_tenant=allow_tenant):
392 raise SafetyError("remote host is not on the official protocols.io allowlist")
393 if not any(parsed.path.startswith(prefix) for prefix in allowed_paths):
394 raise SafetyError("remote URL path is outside the allowlisted API/export paths")
395 return urllib.parse.urlunsplit(
396 ("https", parsed.hostname.lower(), parsed.path, parsed.query, "")
397 )
400def build_url(
401 origin: str,
402 path: str,
403 params: Mapping[str, Any] | None = None,
404) -> str:
405 normalized_origin = validate_origin(origin, allow_tenant=True)
406 if not path.startswith("/") or "\\" in path or "\x00" in path:
407 raise SafetyError("request path must be an absolute URL path")
408 query = urllib.parse.urlencode(params or {}, doseq=True)
409 return urllib.parse.urlunsplit(
410 ("https", urllib.parse.urlsplit(normalized_origin).netloc, path, query, "")
411 )
414def validate_protocol_identifier(value: str) -> str:
415 if _PROTOCOL_ID_RE.fullmatch(value) is None:
416 raise SafetyError(
417 "protocol identifier must be an integer, URI, or DOI with an optional "
418 "/vN or /latest suffix"
419 )
420 return value
423def encode_protocol_identifier(value: str) -> str:
424 validated = validate_protocol_identifier(value)
425 return "/".join(
426 urllib.parse.quote(part, safe="._-") for part in validated.split("/")
427 )
430def validate_guid(value: str) -> str:
431 if _GUID_RE.fullmatch(value) is None:
432 raise SafetyError("GUID must contain exactly 32 hexadecimal characters")
433 return value.upper()
436def validate_nonnegative_integer(value: str, *, name: str, maximum: int) -> int:
437 if _INTEGER_RE.fullmatch(value) is None:
438 raise SafetyError(f"{name} must be an unsigned base-10 integer")
439 parsed = int(value)
440 if parsed > maximum:
441 raise SafetyError(f"{name} must not exceed {maximum}")
442 return parsed
445def credential_status(
446 environ: Mapping[str, str] | None = None,
447) -> dict[str, Any]:
448 """Return presence-only credential status without exposing values."""
450 environment = os.environ if environ is None else environ
451 present = {name: bool(environment.get(name)) for name in CREDENTIAL_ENV_NAMES}
452 return {
453 "named_variables_only": True,
454 "dotenv_loaded": False,
455 "variables": {
456 name: {"present": is_present} for name, is_present in present.items()
457 },
458 "rest_bearer_ready": present[ACCESS_TOKEN_ENV],
459 "oauth_credentials_consumed": False,
460 }
463def _bounded_read(response: Any, max_bytes: int) -> bytes:
464 content_length = response.headers.get("Content-Length")
465 if content_length:
466 try:
467 declared = int(content_length)
468 except (TypeError, ValueError):
469 declared = -1
470 if declared > max_bytes:
471 raise SafetyError(
472 f"response Content-Length exceeds the {max_bytes}-byte safety cap"
473 )
475 chunks: list[bytes] = []
476 total = 0
477 while True:
478 chunk = response.read(min(65_536, max_bytes - total + 1))
479 if not chunk:
480 break
481 total += len(chunk)
482 if total > max_bytes:
483 raise SafetyError(f"response exceeds the {max_bytes}-byte safety cap")
484 chunks.append(chunk)
485 return b"".join(chunks)
488def _retry_delay(headers: Mapping[str, str], attempt: int) -> float:
489 raw = headers.get("Retry-After")
490 if raw is not None:
491 try:
492 seconds = float(raw)
493 except (TypeError, ValueError):
494 seconds = 0.0
495 if math.isfinite(seconds) and seconds > 0:
496 return min(seconds, MAX_RETRY_AFTER_SECONDS)
497 return min(float(2**attempt), MAX_RETRY_AFTER_SECONDS)
500def _api_error_from_body(status: int, body: bytes) -> ApiError:
501 api_status: int | str | None = None
502 if body:
503 try:
504 payload = parse_json_bytes(body, source="error response")
505 except SafetyError:
506 payload = None
507 if isinstance(payload, Mapping):
508 api_status = payload.get("status_code")
509 message = f"protocols.io returned HTTP {status}"
510 if api_status is not None:
511 message += f" with API status {api_status}"
512 return ApiError(message, http_status=status, api_status=api_status)
515def request_bytes(
516 url: str,
517 *,
518 token: str | None,
519 accept: str,
520 timeout: float = DEFAULT_TIMEOUT_SECONDS,
521 retries: int = 1,
522 max_bytes: int = MAX_JSON_RESPONSE_BYTES,
523 opener: Any | None = None,
524 sleep: Callable[[float], None] = time.sleep,
525) -> HttpResult:
526 """Perform one bounded GET request to an allowlisted URL.
528 Only idempotent GET requests are supported. Redirects are rejected. Retries
529 are bounded and limited to 429/500/502/503/504 responses.
530 """
532 safe_url = validate_remote_url(url, allow_tenant=True)
533 timeout = validate_timeout(timeout)
534 retries = validate_retries(retries)
535 if max_bytes < 1 or max_bytes > MAX_PDF_RESPONSE_BYTES:
536 raise SafetyError(f"max_bytes must be between 1 and {MAX_PDF_RESPONSE_BYTES}")
537 if token is not None and (
538 not token
539 or any(character.isspace() or ord(character) < 32 for character in token)
540 or "\x7f" in token
541 ):
542 raise SafetyError("access token is empty or contains forbidden characters")
544 request_headers = {
545 "Accept": accept,
546 "User-Agent": "scientific-agent-skills/protocolsio-integration-1.4",
547 }
548 if token is not None:
549 request_headers["Authorization"] = f"Bearer {token}"
551 transport = opener or urllib.request.build_opener(
552 urllib.request.ProxyHandler({}),
553 NoRedirectHandler(),
554 )
555 retryable = {429, 500, 502, 503, 504}
556 for attempt in range(retries + 1):
557 request = urllib.request.Request(
558 safe_url,
559 headers=request_headers,
560 method="GET",
561 )
562 try:
563 with transport.open(request, timeout=timeout) as response:
564 raw_status = getattr(response, "status", None)
565 if raw_status is None:
566 raw_status = response.getcode()
567 status = int(raw_status)
568 body = _bounded_read(response, max_bytes)
569 headers = {
570 str(key): str(value) for key, value in response.headers.items()
571 }
572 final_url = validate_remote_url(
573 str(getattr(response, "url", safe_url)),
574 allow_tenant=True,
575 )
576 if status >= 400:
577 raise _api_error_from_body(status, body[:MAX_ERROR_BYTES])
578 return HttpResult(
579 status=status,
580 headers=headers,
581 body=body,
582 url=final_url,
583 attempts=attempt + 1,
584 )
585 except urllib.error.HTTPError as exc:
586 headers = {str(key): str(value) for key, value in exc.headers.items()}
587 if exc.code in retryable and attempt < retries:
588 sleep(_retry_delay(headers, attempt))
589 continue
590 try:
591 body = _bounded_read(exc, MAX_ERROR_BYTES)
592 except SafetyError:
593 body = b""
594 raise _api_error_from_body(exc.code, body) from None
595 except urllib.error.URLError as exc:
596 if attempt < retries:
597 sleep(min(float(2**attempt), MAX_RETRY_AFTER_SECONDS))
598 continue
599 reason = clean_text(str(exc.reason), max_chars=200)
600 raise ApiError(f"network request failed: {reason}") from None
602 raise ApiError("bounded request attempts were exhausted")
605def require_api_success(payload: Any) -> Mapping[str, Any]:
606 if not isinstance(payload, Mapping):
607 raise ApiError("protocols.io returned a non-object JSON response")
608 status = payload.get("status_code", 0)
609 if status not in (0, "0", None):
610 raise ApiError(
611 f"protocols.io returned API status {status}",
612 api_status=status,
613 )
614 return payload