Skills › Scientific Agent Skills › labarchive-integration › entry_operations.py
entry_operations.py
386 lines · skills/labarchive-integration/scripts/entry_operations.py at 5f2bc03
1#!/usr/bin/env python3
2"""Offline LabArchives request-signing helpers and redacted request plans.
4The CLI performs no network requests and never prints credentials or reusable
5signatures. Import the functions into institution-reviewed HTTP code when
6needed, and pass returned authentication material directly to the client.
7"""
9from __future__ import annotations
11import argparse
12import base64
13import getpass
14import hashlib
15import hmac
16import json
17import os
18import re
19import sys
20import time
21from collections.abc import Mapping, Sequence
22from typing import Any
23from urllib.parse import quote
25from setup_config import (
26 ENV_ACCESS_KEY_ID,
27 ENV_ACCESS_PASSWORD,
28 ENV_API_URL,
29 ENV_INVENTORY_LAB_ID,
30 ENV_USER_ID,
31 INVENTORY_API_BASE_URL,
32 ConfigError,
33 normalize_eln_api_url,
34)
37_ELN_COMPONENT = re.compile(r"^[a-z][a-z0-9_]*$")
38_INVENTORY_PATH = re.compile(r"^/public/v1/[A-Za-z0-9._~!$&'()*+,;=:@/-]+$")
39_OFFICIAL_VECTOR = {
40 "access_key_id": "0234wedkfjrtfd34er",
41 "api_method_input": "entry_attachment",
42 "expires_ms": 264433207000,
43 "access_password": "1234567890",
44 "signature": (
45 "mT7pS+KgqlNseR0bo4YLQOVIsgOugMWzlQGllInXS25Q7VpA6lRmL0nUq/"
46 "UUdrlF+WV7POYE1vcwvN/pnac7bw=="
47 ),
48}
51def _require_text(value: str, label: str) -> str:
52 normalized = value.strip()
53 if not normalized:
54 raise ConfigError(f"{label} must not be empty")
55 return normalized
58def _require_secret(value: str, label: str) -> str:
59 if not value:
60 raise ConfigError(f"{label} must not be empty")
61 return value
64def _expires_ms(value: int | None = None) -> int:
65 selected = time.time_ns() // 1_000_000 if value is None else value
66 if selected < 0:
67 raise ConfigError("expires must be a non-negative epoch-millisecond value")
68 return selected
71def create_signature(
72 access_key_id: str,
73 api_method_input: str,
74 expires_ms: int,
75 access_password: str,
76) -> str:
77 """Create the documented Base64(HMAC-SHA-512) LabArchives signature.
79 No separators are inserted. The result is not URI-encoded.
80 """
82 key_id = _require_text(access_key_id, "Access Key ID")
83 method_input = _require_text(api_method_input, "API method input")
84 secret = _require_secret(access_password, "Access Password")
85 expires = _expires_ms(expires_ms)
86 message = f"{key_id}{method_input}{expires}".encode("utf-8")
87 digest = hmac.new(secret.encode("utf-8"), message, hashlib.sha512).digest()
88 return base64.b64encode(digest).decode("ascii")
91def encode_eln_signature(signature: str) -> str:
92 """URI-encode a Base64 signature for an ELN sig query parameter."""
94 return quote(_require_text(signature, "signature"), safe="")
97def build_eln_auth_params(
98 access_key_id: str,
99 access_password: str,
100 api_method: str,
101 expires_ms: int | None = None,
102) -> dict[str, str]:
103 """Return ELN query authentication parameters.
105 Keep the returned mapping out of logs. A normal query encoder will
106 URI-encode the Base64 signature.
107 """
109 method = validate_eln_component(api_method, "ELN API method")
110 expires = _expires_ms(expires_ms)
111 signature = create_signature(access_key_id, method, expires, access_password)
112 return {
113 "akid": _require_text(access_key_id, "Access Key ID"),
114 "expires": str(expires),
115 "sig": signature,
116 }
119def build_inventory_headers(
120 access_key_id: str,
121 access_password: str,
122 user_id: str,
123 lab_id: str | None,
124 relative_path: str,
125 expires_ms: int | None = None,
126) -> dict[str, str]:
127 """Return documented Inventory v1 authentication headers.
129 The relative path includes resolved route parameters and excludes any query
130 string. The users/me bootstrap route does not require a Lab ID; pass None
131 or an empty string there to omit that header. Other routes require a Lab ID.
132 Keep the returned mapping out of logs.
133 """
135 path = validate_inventory_path(relative_path)
136 expires = _expires_ms(expires_ms)
137 signature = create_signature(access_key_id, path, expires, access_password)
138 headers = {
139 "X-LabArchives-UId": _require_text(user_id, "Inventory user ID"),
140 "X-LabArchives-AKId": _require_text(access_key_id, "Access Key ID"),
141 "X-LabArchives-Signature": signature,
142 "X-LabArchives-Expires": str(expires),
143 }
144 if path != "/public/v1/users/me" or (lab_id and lab_id.strip()):
145 headers["X-LabArchives-LabId"] = _require_text(lab_id or "", "Inventory Lab ID")
146 return headers
149def validate_eln_component(value: str, label: str) -> str:
150 """Validate a class or method copied from the official ELN API tree."""
152 normalized = value.strip()
153 if not _ELN_COMPONENT.fullmatch(normalized):
154 raise ConfigError(
155 f"{label} must match {_ELN_COMPONENT.pattern!r}; "
156 "copy it from the current official method page"
157 )
158 return normalized
161def validate_inventory_path(value: str) -> str:
162 """Validate an exact, resolved Inventory v1 relative route."""
164 path = value.strip()
165 if "?" in path or "#" in path:
166 raise ConfigError(
167 "Inventory signature path must exclude query strings and fragments"
168 )
169 if "\\" in path or "%" in path or any(character.isspace() for character in path):
170 raise ConfigError(
171 "Inventory signature path must be an unencoded relative route"
172 )
173 if "{" in path or "}" in path:
174 raise ConfigError("resolve all Inventory route placeholders before signing")
175 if not _INVENTORY_PATH.fullmatch(path):
176 raise ConfigError(
177 "Inventory path must be an exact route beginning with /public/v1/"
178 )
179 segments = path.split("/")
180 if any(segment in {".", ".."} for segment in segments) or "//" in path:
181 raise ConfigError("Inventory path contains an unsafe or ambiguous segment")
182 return path
185def _secret_fingerprint(value: str) -> str:
186 return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
189def _dump(payload: Mapping[str, Any], *, compact: bool, stream: Any) -> None:
190 if compact:
191 json.dump(payload, stream, sort_keys=True, separators=(",", ":"))
192 else:
193 json.dump(payload, stream, indent=2, sort_keys=True)
194 stream.write("\n")
197def _load_credentials(
198 args: argparse.Namespace, env: Mapping[str, str]
199) -> tuple[str, str]:
200 access_key_id = env.get(ENV_ACCESS_KEY_ID, "").strip()
201 if not access_key_id:
202 raise ConfigError(
203 f"required environment variable is missing: {ENV_ACCESS_KEY_ID}"
204 )
206 access_password = env.get(ENV_ACCESS_PASSWORD, "")
207 if not access_password and args.prompt_missing_secret:
208 access_password = getpass.getpass(f"{ENV_ACCESS_PASSWORD}: ")
209 if not access_password:
210 raise ConfigError(
211 f"required environment variable is missing: {ENV_ACCESS_PASSWORD}"
212 )
213 return access_key_id, access_password
216def command_self_test(args: argparse.Namespace) -> int:
217 actual = create_signature(
218 _OFFICIAL_VECTOR["access_key_id"],
219 _OFFICIAL_VECTOR["api_method_input"],
220 int(_OFFICIAL_VECTOR["expires_ms"]),
221 _OFFICIAL_VECTOR["access_password"],
222 )
223 passed = hmac.compare_digest(actual, str(_OFFICIAL_VECTOR["signature"]))
224 payload = {
225 "passed": passed,
226 "source": "official LabArchives Call Authentication example",
227 "signature_fingerprint": _secret_fingerprint(actual),
228 "credentials_used": "public dummy test vector only",
229 "remote_request_performed": False,
230 }
231 _dump(payload, compact=args.compact, stream=sys.stdout)
232 return 0 if passed else 1
235def command_eln_plan(
236 args: argparse.Namespace, env: Mapping[str, str] | None = None
237) -> int:
238 selected_env = os.environ if env is None else env
239 access_key_id, access_password = _load_credentials(args, selected_env)
240 api_class = validate_eln_component(args.api_class, "ELN API class")
241 api_method = validate_eln_component(args.api_method, "ELN API method")
242 api_url = normalize_eln_api_url(args.api_url or selected_env.get(ENV_API_URL, ""))
243 auth = build_eln_auth_params(
244 access_key_id,
245 access_password,
246 api_method,
247 expires_ms=args.expires_ms,
248 )
249 payload = {
250 "dry_run": True,
251 "remote_request_performed": False,
252 "api_surface": "legacy ELN",
253 "endpoint_without_query": f"{api_url}/{api_class}/{api_method}",
254 "signature_input": api_method,
255 "expires_ms": int(auth["expires"]),
256 "authentication_parameter_names": sorted(auth),
257 "access_key_id_fingerprint": _secret_fingerprint(access_key_id),
258 "signature_fingerprint": _secret_fingerprint(auth["sig"]),
259 "reusable_authentication_material_printed": False,
260 "warning": (
261 "Verify the exact official method page, HTTP verb, and parameters "
262 "before implementing a request."
263 ),
264 }
265 _dump(payload, compact=args.compact, stream=sys.stdout)
266 return 0
269def command_inventory_plan(
270 args: argparse.Namespace, env: Mapping[str, str] | None = None
271) -> int:
272 selected_env = os.environ if env is None else env
273 access_key_id, access_password = _load_credentials(args, selected_env)
274 user_id = selected_env.get(ENV_USER_ID, "")
275 lab_id = selected_env.get(ENV_INVENTORY_LAB_ID, "")
276 headers = build_inventory_headers(
277 access_key_id,
278 access_password,
279 user_id,
280 lab_id,
281 args.path,
282 expires_ms=args.expires_ms,
283 )
284 payload = {
285 "dry_run": True,
286 "remote_request_performed": False,
287 "api_surface": "Inventory API v1",
288 "relative_path": validate_inventory_path(args.path),
289 "absolute_base_url": INVENTORY_API_BASE_URL,
290 "expires_ms": int(headers["X-LabArchives-Expires"]),
291 "authentication_header_names": sorted(headers),
292 "access_key_id_fingerprint": _secret_fingerprint(access_key_id),
293 "signature_fingerprint": _secret_fingerprint(
294 headers["X-LabArchives-Signature"]
295 ),
296 "reusable_authentication_material_printed": False,
297 "warning": (
298 "The Inventory overview documents this base URL. Confirm the account "
299 "deployment before a live request; do not infer regional variants."
300 ),
301 }
302 _dump(payload, compact=args.compact, stream=sys.stdout)
303 return 0
306def _add_plan_options(parser: argparse.ArgumentParser) -> None:
307 parser.add_argument(
308 "--expires-ms",
309 type=int,
310 help=(
311 "explicit current epoch milliseconds for deterministic validation; "
312 "defaults to the local current time"
313 ),
314 )
315 parser.add_argument(
316 "--prompt-missing-secret",
317 action="store_true",
318 help="use getpass for a missing Access Password without saving it",
319 )
320 parser.add_argument("--compact", action="store_true", help="emit compact JSON")
323def build_parser() -> argparse.ArgumentParser:
324 parser = argparse.ArgumentParser(
325 description=(
326 "Generate redacted, offline LabArchives signing plans. "
327 "No HTTP requests are implemented."
328 )
329 )
330 subparsers = parser.add_subparsers(dest="command", required=True)
332 self_test = subparsers.add_parser(
333 "self-test", help="check HMAC implementation against the official vector"
334 )
335 self_test.add_argument("--compact", action="store_true", help="emit compact JSON")
336 self_test.set_defaults(handler=command_self_test)
338 eln = subparsers.add_parser(
339 "eln-plan", help="create a redacted legacy ELN request-signing plan"
340 )
341 eln.add_argument(
342 "--api-url",
343 help=f"allowlisted ELN API URL; otherwise read {ENV_API_URL}",
344 )
345 eln.add_argument("--api-class", required=True, help="official ELN API class")
346 eln.add_argument("--api-method", required=True, help="official ELN API method")
347 _add_plan_options(eln)
348 eln.set_defaults(handler=command_eln_plan)
350 inventory = subparsers.add_parser(
351 "inventory-plan",
352 help="create a redacted Inventory v1 request-signing plan",
353 )
354 inventory.add_argument(
355 "--path",
356 required=True,
357 help="exact /public/v1/... route with path values resolved and no query",
358 )
359 _add_plan_options(inventory)
360 inventory.set_defaults(handler=command_inventory_plan)
361 return parser
364def main(argv: Sequence[str] | None = None) -> int:
365 parser = build_parser()
366 args = parser.parse_args(argv)
367 try:
368 return int(args.handler(args))
369 except ConfigError as exc:
370 _dump(
371 {
372 "valid": False,
373 "error": str(exc),
374 "remote_request_performed": False,
375 },
376 compact=getattr(args, "compact", False),
377 stream=sys.stderr,
378 )
379 return 2
380 except KeyboardInterrupt:
381 print('{"valid":false,"error":"cancelled"}', file=sys.stderr)
382 return 130
385if __name__ == "__main__":
386 raise SystemExit(main())