Skills › Scientific Agent Skills › omero-integration › validate_config.py
validate_config.py
140 lines · skills/omero-integration/scripts/validate_config.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate named OMERO endpoint/auth variables without contacting OMERO."""
4from __future__ import annotations
6import argparse
7import json
8import socket
9import sys
10from typing import Any
12from omero_common import (
13 ConfigError,
14 config_summary,
15 load_connection_config,
16 scrubbed_error,
17)
20def build_parser() -> argparse.ArgumentParser:
21 parser = argparse.ArgumentParser(
22 description=(
23 "Validate only named OMERO_* variables locally. By default this "
24 "does not resolve DNS and never contacts an OMERO server."
25 )
26 )
27 parser.add_argument(
28 "--require-auth",
29 action="store_true",
30 help=(
31 "Require OMERO_SESSION_KEY or both OMERO_USER and "
32 "OMERO_PASSWORD."
33 ),
34 )
35 parser.add_argument(
36 "--resolve-host",
37 action="store_true",
38 help=(
39 "Resolve OMERO_HOST through DNS without opening an OMERO "
40 "connection."
41 ),
42 )
43 parser.add_argument(
44 "--json",
45 action="store_true",
46 help="Emit machine-readable JSON instead of a concise text summary.",
47 )
48 return parser
51def resolve_host(host: str, port: int, *, limit: int = 10) -> list[str]:
52 """Resolve a bounded set of unique numeric addresses."""
54 addresses: list[str] = []
55 seen: set[str] = set()
56 for result in socket.getaddrinfo(
57 host,
58 port,
59 type=socket.SOCK_STREAM,
60 ):
61 address = result[4][0]
62 if address not in seen:
63 seen.add(address)
64 addresses.append(address)
65 if len(addresses) >= limit:
66 break
67 return addresses
70def format_text(payload: dict[str, Any]) -> str:
71 endpoint = payload["endpoint"]
72 lines = [
73 "OMERO configuration is valid.",
74 f"Endpoint: {endpoint['host']}:{endpoint['port']}",
75 f"Secure transport requested: {endpoint['secure_transport']}",
76 f"Authentication mode: {payload['authentication_mode']}",
77 "Credential values included: false",
78 ]
79 if "resolved_addresses" in payload:
80 lines.append(
81 "Resolved addresses: "
82 + ", ".join(payload["resolved_addresses"])
83 )
84 for warning in payload.get("warnings", []):
85 lines.append(f"Warning: {warning}")
86 return "\n".join(lines)
89def main(argv: list[str] | None = None) -> int:
90 args = build_parser().parse_args(argv)
91 try:
92 config = load_connection_config(require_auth=args.require_auth)
93 payload = config_summary(config)
94 payload["status"] = "valid"
95 payload["server_contacted"] = False
96 payload["warnings"] = []
97 if not config.secure:
98 payload["warnings"].append(
99 "OMERO_SECURE is false; post-login traffic may be unencrypted"
100 )
101 if config.authentication_mode == "missing":
102 payload["warnings"].append(
103 "No complete authentication credential is configured"
104 )
105 if args.resolve_host:
106 payload["resolved_addresses"] = resolve_host(
107 config.host,
108 config.port,
109 )
110 payload["dns_resolution_performed"] = True
111 else:
112 payload["dns_resolution_performed"] = False
114 if args.json:
115 print(json.dumps(payload, indent=2, sort_keys=True))
116 else:
117 print(format_text(payload))
118 return 0
119 except (ConfigError, socket.gaierror, OSError) as error:
120 if args.json:
121 print(
122 json.dumps(
123 {
124 "status": "invalid",
125 "server_contacted": False,
126 "error": scrubbed_error(error),
127 "credential_values_included": False,
128 },
129 indent=2,
130 sort_keys=True,
131 ),
132 file=sys.stderr,
133 )
134 else:
135 print(scrubbed_error(error), file=sys.stderr)
136 return 2
139if __name__ == "__main__":
140 raise SystemExit(main())