Skills › Scientific Agent Skills › omero-integration › omero_common.py
omero_common.py
493 lines · skills/omero-integration/scripts/omero_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared safety utilities for the bundled OMERO client helpers."""
4from __future__ import annotations
6import argparse
7import contextlib
8import json
9import math
10import os
11import tempfile
12from collections.abc import Iterable, Iterator, Mapping
13from dataclasses import dataclass
14from datetime import date, datetime
15from itertools import islice
16from pathlib import Path
17from typing import Any
20NAMED_ENV_VARS = (
21 "OMERO_HOST",
22 "OMERO_PORT",
23 "OMERO_USER",
24 "OMERO_PASSWORD",
25 "OMERO_SESSION_KEY",
26 "OMERO_SECURE",
27)
29TRUE_VALUES = frozenset({"1", "true", "yes", "on"})
30FALSE_VALUES = frozenset({"0", "false", "no", "off"})
33class ConfigError(ValueError):
34 """Raised when named OMERO configuration is invalid."""
37class OutputPathError(ValueError):
38 """Raised when an output path is unsafe or ambiguous."""
41class DependencyError(RuntimeError):
42 """Raised when an optional runtime dependency is unavailable."""
45@dataclass(frozen=True)
46class ConnectionConfig:
47 """Validated connection material. Never serialize this dataclass."""
49 host: str
50 port: int
51 secure: bool
52 username: str | None
53 password: str | None
54 session_key: str | None
56 @property
57 def authentication_mode(self) -> str:
58 if self.session_key:
59 return "session_key"
60 if self.username and self.password:
61 return "username_password"
62 return "missing"
65@dataclass(frozen=True)
66class BoundedResult:
67 """A bounded materialization with explicit truncation state."""
69 items: list[Any]
70 truncated: bool
73def parse_bool(value: str, *, variable: str) -> bool:
74 """Parse a strict environment boolean."""
76 normalized = value.strip().lower()
77 if normalized in TRUE_VALUES:
78 return True
79 if normalized in FALSE_VALUES:
80 return False
81 allowed = ", ".join(sorted(TRUE_VALUES | FALSE_VALUES))
82 raise ConfigError(f"{variable} must be one of: {allowed}")
85def validate_host(value: str) -> str:
86 """Validate a hostname without resolving or connecting."""
88 host = value.strip()
89 if not host:
90 raise ConfigError("OMERO_HOST is required")
91 if "\x00" in host:
92 raise ConfigError("OMERO_HOST contains a NUL byte")
93 if any(character.isspace() for character in host):
94 raise ConfigError("OMERO_HOST must not contain whitespace")
95 if "://" in host:
96 raise ConfigError("OMERO_HOST must be a hostname, not a URL")
97 if "/" in host or "\\" in host:
98 raise ConfigError("OMERO_HOST must not contain a path")
99 if host.startswith("-"):
100 raise ConfigError("OMERO_HOST must not begin with '-'")
101 if len(host) > 255:
102 raise ConfigError("OMERO_HOST is too long")
103 return host
106def parse_port(value: str | None) -> int:
107 """Parse OMERO_PORT with the documented SSL-router default."""
109 raw = "4064" if value is None or not value.strip() else value.strip()
110 try:
111 port = int(raw, 10)
112 except ValueError as exc:
113 raise ConfigError("OMERO_PORT must be an integer") from exc
114 if not 1 <= port <= 65535:
115 raise ConfigError("OMERO_PORT must be between 1 and 65535")
116 return port
119def load_connection_config(
120 *,
121 require_auth: bool,
122 environ: Mapping[str, str] | None = None,
123) -> ConnectionConfig:
124 """Read only the named OMERO_* variables and validate them."""
126 source = os.environ if environ is None else environ
127 values = {name: source.get(name) for name in NAMED_ENV_VARS}
129 host = validate_host(values["OMERO_HOST"] or "")
130 port = parse_port(values["OMERO_PORT"])
131 secure_raw = values["OMERO_SECURE"]
132 secure = True if secure_raw is None else parse_bool(
133 secure_raw,
134 variable="OMERO_SECURE",
135 )
137 username = (values["OMERO_USER"] or "").strip() or None
138 password = values["OMERO_PASSWORD"]
139 if password == "":
140 password = None
141 session_key = (values["OMERO_SESSION_KEY"] or "").strip() or None
143 # An explicit session is the sole credential in use. Stale password
144 # variables are deliberately ignored and never surfaced.
145 if session_key:
146 username = None
147 password = None
148 elif bool(username) != bool(password):
149 raise ConfigError(
150 "OMERO_USER and OMERO_PASSWORD must be set together, "
151 "or use OMERO_SESSION_KEY"
152 )
154 config = ConnectionConfig(
155 host=host,
156 port=port,
157 secure=secure,
158 username=username,
159 password=password,
160 session_key=session_key,
161 )
162 if require_auth and config.authentication_mode == "missing":
163 raise ConfigError(
164 "Authentication requires OMERO_SESSION_KEY or both "
165 "OMERO_USER and OMERO_PASSWORD"
166 )
167 return config
170def config_summary(
171 config: ConnectionConfig,
172 *,
173 environ: Mapping[str, str] | None = None,
174) -> dict[str, Any]:
175 """Return a credential-safe configuration summary."""
177 source = os.environ if environ is None else environ
178 return {
179 "endpoint": {
180 "host": config.host,
181 "port": config.port,
182 "secure_transport": config.secure,
183 },
184 "authentication_mode": config.authentication_mode,
185 "named_variables_present": {
186 name: bool(source.get(name))
187 for name in NAMED_ENV_VARS
188 },
189 "credential_values_included": False,
190 }
193def require_secure_transport(
194 config: ConnectionConfig,
195 *,
196 allow_insecure_transport: bool,
197) -> None:
198 """Refuse post-login cleartext unless explicitly overridden."""
200 if not config.secure and not allow_insecure_transport:
201 raise ConfigError(
202 "OMERO_SECURE is false; pass --allow-insecure-transport only "
203 "after reviewing the server transport policy"
204 )
207@contextlib.contextmanager
208def gateway_session(
209 config: ConnectionConfig,
210 *,
211 allow_insecure_transport: bool,
212) -> Iterator[Any]:
213 """Open and always close a BlitzGateway connection."""
215 require_secure_transport(
216 config,
217 allow_insecure_transport=allow_insecure_transport,
218 )
219 if config.authentication_mode == "missing":
220 raise ConfigError("Remote execution requires authentication")
222 try:
223 from omero.gateway import BlitzGateway
224 except ImportError as exc:
225 raise DependencyError(
226 "omero-py and a matching ZeroC IcePy wheel are required "
227 "for --execute"
228 ) from exc
230 connection = None
231 try:
232 if config.session_key:
233 connection = BlitzGateway(
234 host=config.host,
235 port=config.port,
236 secure=config.secure,
237 )
238 connected = connection.connect(sUuid=config.session_key)
239 else:
240 connection = BlitzGateway(
241 config.username,
242 config.password,
243 host=config.host,
244 port=config.port,
245 secure=config.secure,
246 )
247 connected = connection.connect()
249 if not connected:
250 raise RuntimeError("OMERO connection failed")
251 yield connection
252 finally:
253 if connection is not None:
254 with contextlib.suppress(Exception):
255 # A joined session may still be used by its original client.
256 # BlitzGateway.close() defaults to killing it regardless of
257 # the server's connection reference count.
258 connection.close(hard=not bool(config.session_key))
261def take_bounded(iterable: Iterable[Any], limit: int) -> BoundedResult:
262 """Materialize at most limit values and detect one additional value."""
264 if limit < 0:
265 raise ValueError("limit must be non-negative")
266 items = list(islice(iterable, limit + 1))
267 return BoundedResult(items=items[:limit], truncated=len(items) > limit)
270def unwrap_omero(value: Any) -> Any:
271 """Unwrap a generated OMERO rtype when possible."""
273 if value is None:
274 return None
275 getter = getattr(value, "getValue", None)
276 if callable(getter):
277 return getter()
278 if hasattr(value, "val"):
279 return value.val
280 return value
283def json_safe(
284 value: Any,
285 *,
286 max_string_length: int = 512,
287 max_collection_length: int = 1000,
288 _depth: int = 0,
289) -> Any:
290 """Convert known values to bounded JSON without unsafe repr output."""
292 if _depth > 8:
293 return {"omitted": "maximum nesting depth exceeded"}
294 if value is None or isinstance(value, (bool, int)):
295 return value
296 if isinstance(value, float):
297 if math.isfinite(value):
298 return value
299 return {"non_finite_float": str(value)}
300 if isinstance(value, str):
301 if len(value) <= max_string_length:
302 return value
303 return value[:max_string_length] + "…[truncated]"
304 if isinstance(value, bytes):
305 return {"bytes_omitted": len(value)}
306 if isinstance(value, (date, datetime)):
307 return value.isoformat()
309 unwrapped = unwrap_omero(value)
310 if unwrapped is not value:
311 return json_safe(
312 unwrapped,
313 max_string_length=max_string_length,
314 max_collection_length=max_collection_length,
315 _depth=_depth + 1,
316 )
317 if isinstance(value, Mapping):
318 total_items = len(value)
319 converted = {
320 str(key)[:max_string_length]: json_safe(
321 item,
322 max_string_length=max_string_length,
323 max_collection_length=max_collection_length,
324 _depth=_depth + 1,
325 )
326 for key, item in islice(
327 value.items(),
328 max_collection_length,
329 )
330 }
331 if total_items > max_collection_length:
332 converted["__truncated_items__"] = (
333 total_items - max_collection_length
334 )
335 return converted
336 if isinstance(value, (list, tuple)):
337 converted = [
338 json_safe(
339 item,
340 max_string_length=max_string_length,
341 max_collection_length=max_collection_length,
342 _depth=_depth + 1,
343 )
344 for item in value[:max_collection_length]
345 ]
346 if len(value) > max_collection_length:
347 converted.append(
348 {
349 "truncated_items": (
350 len(value) - max_collection_length
351 )
352 }
353 )
354 return converted
355 return {"unsupported_type": type(value).__name__}
358def positive_int(value: str) -> int:
359 """Argparse-compatible positive integer parser."""
361 try:
362 parsed = int(value, 10)
363 except ValueError as exc:
364 raise ValueError("expected an integer") from exc
365 if parsed <= 0:
366 raise ValueError("expected a positive integer")
367 return parsed
370def bounded_int(
371 value: int,
372 *,
373 name: str,
374 minimum: int,
375 maximum: int,
376) -> int:
377 """Validate an integer bound with a stable message."""
379 if not minimum <= value <= maximum:
380 raise ValueError(
381 f"{name} must be between {minimum} and {maximum}"
382 )
383 return value
386def atomic_write_json(
387 output: str | os.PathLike[str],
388 payload: Any,
389 *,
390 overwrite: bool,
391) -> Path:
392 """Atomically write JSON to an existing directory with mode 0600."""
394 requested = Path(output).expanduser()
395 if requested.name in {"", ".", ".."}:
396 raise OutputPathError("output must name a JSON file")
397 if requested.suffix.lower() != ".json":
398 raise OutputPathError("output filename must end in .json")
399 if requested.is_symlink():
400 raise OutputPathError("refusing to write through an output symlink")
402 try:
403 parent = requested.parent.resolve(strict=True)
404 except FileNotFoundError as exc:
405 raise OutputPathError("output parent directory does not exist") from exc
406 if not parent.is_dir():
407 raise OutputPathError("output parent is not a directory")
409 target = parent / requested.name
410 if target.is_symlink():
411 raise OutputPathError("refusing to replace an output symlink")
412 if target.exists():
413 if not target.is_file():
414 raise OutputPathError("output exists and is not a regular file")
415 if not overwrite:
416 raise FileExistsError(f"output already exists: {target}")
418 descriptor, temporary_name = tempfile.mkstemp(
419 prefix=f".{target.name}.",
420 suffix=".tmp",
421 dir=parent,
422 )
423 temporary = Path(temporary_name)
424 try:
425 os.fchmod(descriptor, 0o600)
426 with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
427 descriptor = -1
428 json.dump(
429 payload,
430 handle,
431 indent=2,
432 sort_keys=True,
433 ensure_ascii=False,
434 )
435 handle.write("\n")
436 handle.flush()
437 os.fsync(handle.fileno())
438 os.replace(temporary, target)
439 os.chmod(target, 0o600)
440 except Exception:
441 if descriptor >= 0:
442 os.close(descriptor)
443 with contextlib.suppress(FileNotFoundError):
444 temporary.unlink()
445 raise
446 return target
449def emit_json(
450 payload: Any,
451 *,
452 output: str | os.PathLike[str] | None,
453 overwrite: bool,
454) -> Path | None:
455 """Write to a safe file or emit bounded JSON to stdout."""
457 if output is None:
458 print(
459 json.dumps(
460 payload,
461 indent=2,
462 sort_keys=True,
463 ensure_ascii=False,
464 )
465 )
466 return None
467 return atomic_write_json(output, payload, overwrite=overwrite)
470def scrubbed_error(error: BaseException) -> str:
471 """Describe an error without serializing its potentially sensitive text."""
473 return (
474 f"{type(error).__name__}: operation failed; "
475 "credential values were not logged"
476 )
479def main(argv: list[str] | None = None) -> int:
480 """Expose module documentation without loading optional OMERO packages."""
482 parser = argparse.ArgumentParser(
483 description=(
484 "Shared local safety utilities for the bundled OMERO helpers. "
485 "This module does not connect to OMERO when invoked directly."
486 )
487 )
488 parser.parse_args(argv)
489 return 0
492if __name__ == "__main__":
493 raise SystemExit(main())