From 25162b8a45cf995a7e2c650ad84363ed67e78263 Mon Sep 17 00:00:00 2001 From: DanailDR Date: Tue, 29 Sep 2026 13:08:58 +0300 Subject: [PATCH] Add web service inspection report --- README.md | 7 + Web/inspect-web-services.md | 65 ++++ Web/inspect-web-services.sh | 604 ++++++++++++++++++++++++++++++++++++ 3 files changed, 676 insertions(+) create mode 100644 Web/inspect-web-services.md create mode 100755 Web/inspect-web-services.sh diff --git a/README.md b/README.md index 42c41a7..8b7fbce 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,8 @@ strixscripts/ │ └── system-report.sh моментен системен отчет с TXT/JSON/HTML export ├── Virtualisation/ │ └── virtualisation-check.sh проверка на инсталирани виртуализационни решения и инстанции +├── Web/ +│ └── inspect-web-services.sh уеб сървъри, proxy, vhost и listeners на 80/443 └── <категория>/ функционална група скриптове (предстои) ``` @@ -67,3 +69,8 @@ strixscripts/ - [`virtualisation-check.sh`](./Virtualisation/virtualisation-check.md) — открива VM/container решения и отчита инсталираните и работещите им инстанции. + +### Web/ + +- [`inspect-web-services.sh`](./Web/inspect-web-services.md) — открива web + servers/reverse proxies, слушатели на портове 80/443 и Apache/Nginx vhosts. diff --git a/Web/inspect-web-services.md b/Web/inspect-web-services.md new file mode 100644 index 0000000..1ac972c --- /dev/null +++ b/Web/inspect-web-services.md @@ -0,0 +1,65 @@ +# inspect-web-services.sh + +Открива инсталирани web server и reverse-proxy решения, проверява слушателите +на портове 80/443 и извежда Apache/Nginx virtual host конфигурациите и +активните site symlink-и. Скриптът е read-only и не променя услуги или firewall. + +## Какво прави + +Проверява TCP и UDP слушатели на портове 80 и 443 чрез `ss`, включително +процес/PID, когато правата на текущия потребител позволяват това. Така се +откриват и ad-hoc процеси като Python/Node web server, дори да не са сред +известните инсталирани решения. При липсващи process details в отчета се +предлага повторно изпълнение със `sudo`. + +Търси Apache HTTP Server, Nginx, Caddy, HAProxy, Traefik, Lighttpd, Envoy, +Varnish и OpenResty чрез изпълними файлове, package databases и systemd unit-и. +При Apache/Nginx отчита hostname/ServerName, listen address/port, document root, +пътя към конфигурацията, symlink-а от `sites-enabled` и URL адресите, които +могат да се отворят в браузър. URL-ите са изведени като линкове в HTML report. + +Услуга се показва като инсталирана независимо дали в момента работи; състоянието +на systemd unit-а и активните port listeners се показват отделно. + +## Изисквания + +- Linux, Bash и Python 3 +- `ss` е задължителен за откриването на слушатели +- `cat` е задължителен за предаване на Python helper-а, `python3` за анализ и export +- `systemctl`, `dpkg-query`, `rpm`, `apk`, `lsof`, `apache2ctl`/`apachectl`, + `httpd` и `nginx` са незадължителни; липсата им ограничава само съответните + допълнителни сведения +- За пълни PID/process command line данни може да са нужни root права + +## Употреба + +```bash +./Web/inspect-web-services.sh +./Web/inspect-web-services.sh --help +sudo ./Web/inspect-web-services.sh +./Web/inspect-web-services.sh --output "$HOME/web-services.json" --format json +./Web/inspect-web-services.sh --output /tmp/web-services.html --force +``` + +| Опция | Описание | +|-------|----------| +| `-o`, `--output PATH` | Записва директно във файл; без тази опция предлага интерактивен запис | +| `-f`, `--format FORMAT` | Формат `txt`, `json` или `html` | +| `--force` | Позволява презапис на съществуващ файл без потвърждение | +| `-h`, `--help` | Показва помощта и приключва | + +Без зададен `--format` форматът се определя от разширението на `--output`, +иначе се използва TXT. При интерактивен запис началният път е +`~/web-services-report.`. В non-interactive режим без `--output` +отчетът се извежда към stdout; `--format` определя сериализацията. + +## Ограничения + +Apache vhost информацията се чете от конфигурационните файлове в +`/etc/apache2/sites-enabled` или `/etc/httpd/conf.d`; `apache2ctl -S` се използва +като fallback. Nginx се анализира от `nginx -T`, което може да изисква права за +прочит на включените конфигурации. Скриптът открива site symlink-и, но не +проверява дали DNS записите сочат към тази машина, дали TLS сертификатът е +валиден или дали HTTP отговорът на всеки virtual host е успешен. Package +detection зависи от наличния package manager; процесът на произволен listener +се показва дори когато приложението му не е разпознато като познат сървър. \ No newline at end of file diff --git a/Web/inspect-web-services.sh b/Web/inspect-web-services.sh new file mode 100755 index 0000000..2a14f90 --- /dev/null +++ b/Web/inspect-web-services.sh @@ -0,0 +1,604 @@ +#!/usr/bin/env bash +# +# inspect-web-services.sh — inspect web servers, reverse proxies, and port 80/443 listeners. +# Full documentation: see inspect-web-services.md in this same folder. + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +# shellcheck source=lib/common.sh +source "${SCRIPT_DIR}/../lib/common.sh" + +SCRIPT_HELP=$(cat <<'HELP' +Usage: ./Web/inspect-web-services.sh [OPTIONS] + +Detects installed web servers and reverse proxies, checks TCP/UDP listeners on +ports 80 and 443, and reports Apache/Nginx virtual hosts, URLs, and enabled-site +configuration links. It does not change service or firewall settings. + +Options: + -o, --output PATH Save directly to PATH; otherwise ask interactively + -f, --format FORMAT Report format: txt, json, or html + --force Overwrite an existing output file without prompting + -h, --help Show this help message and exit + +Examples: + ./Web/inspect-web-services.sh + ./Web/inspect-web-services.sh --output "$HOME/web-services.json" --format json + sudo ./Web/inspect-web-services.sh +HELP +) + +common::show_help_if_requested "$@" + +OUTPUT_PATH="" +OUTPUT_FORMAT="" +FORCE=0 +while [[ $# -gt 0 ]]; do + case "$1" in + -o|--output) + [[ $# -ge 2 && -n "$2" ]] || common::die "$1 requires a path" + OUTPUT_PATH="$2" + shift 2 + ;; + -f|--format) + [[ $# -ge 2 ]] || common::die "$1 requires txt, json, or html" + OUTPUT_FORMAT="${2,,}" + case "$OUTPUT_FORMAT" in + txt|json|html) ;; + *) common::die "unsupported report format: $2 (use txt, json, or html)" ;; + esac + shift 2 + ;; + --force) + FORCE=1 + shift + ;; + *) + common::die "unknown argument: $1" + ;; + esac +done + +common::require_cmd python3 +common::require_cmd cat +common::require_cmd ss + +python3 -c "$(cat <<'PYCODE' +import datetime +import html +import json +import os +import pathlib +import re +import shutil +import subprocess +import sys +import tempfile + + +SERVER_DEFINITIONS = [ + {"name": "Apache HTTP Server", "binaries": ["apache2ctl", "apachectl", "httpd", "apache2"], "packages": [r"^apache2(?:-|$)", r"^httpd(?:-|$)"], "units": ["apache2.service", "httpd.service"], "ports": [80, 443]}, + {"name": "Nginx", "binaries": ["nginx"], "packages": [r"^nginx(?:-|$)"], "units": ["nginx.service"], "ports": [80, 443]}, + {"name": "Caddy", "binaries": ["caddy"], "packages": [r"^caddy(?:-|$)"], "units": ["caddy.service"], "ports": [80, 443]}, + {"name": "HAProxy", "binaries": ["haproxy"], "packages": [r"^haproxy(?:-|$)"], "units": ["haproxy.service"], "ports": [80, 443]}, + {"name": "Traefik", "binaries": ["traefik"], "packages": [r"^traefik(?:-|$)"], "units": ["traefik.service"], "ports": [80, 443]}, + {"name": "Lighttpd", "binaries": ["lighttpd"], "packages": [r"^lighttpd(?:-|$)"], "units": ["lighttpd.service"], "ports": [80, 443]}, + {"name": "Envoy", "binaries": ["envoy"], "packages": [r"^envoy(?:-|$)"], "units": ["envoy.service"], "ports": [80, 443]}, + {"name": "Varnish", "binaries": ["varnishd"], "packages": [r"^varnish(?:-|$)"], "units": ["varnish.service"], "ports": [80, 443]}, + {"name": "OpenResty", "binaries": ["openresty"], "packages": [r"^openresty(?:-|$)"], "units": ["openresty.service"], "ports": [80, 443]}, +] +SEARCH_PATHS = ("/usr/sbin", "/usr/bin", "/usr/local/sbin", "/usr/local/bin", "/sbin", "/bin") + + +def run(command, timeout=15): + try: + result = subprocess.run( + command, + check=False, + capture_output=True, + text=True, + errors="replace", + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as error: + return None, str(error) + if result.returncode != 0: + return result.stdout.strip() or None, result.stderr.strip() or f"exit status {result.returncode}" + return result.stdout.strip(), None + + +def find_binary(names): + for name in names: + path = shutil.which(name) + if path: + return path + for directory in SEARCH_PATHS: + candidate = pathlib.Path(directory) / name + if candidate.is_file() and os.access(candidate, os.X_OK): + return str(candidate) + return None + + +def installed_packages(): + packages = set() + dpkg = find_binary(["dpkg-query"]) + if dpkg: + output, _ = run([dpkg, "-W", "-f=${db:Status-Abbrev} ${binary:Package}\n"]) + if output: + packages.update(line.split(maxsplit=1)[1] for line in output.splitlines() if line.startswith("ii ") and len(line.split(maxsplit=1)) == 2) + rpm = find_binary(["rpm"]) + if rpm: + output, _ = run([rpm, "-qa", "--qf", "%{NAME}\n"]) + if output: + packages.update(output.splitlines()) + apk = find_binary(["apk"]) + if apk: + output, _ = run([apk, "info"]) + if output: + packages.update(output.splitlines()) + return packages + + +def package_matches(packages, patterns): + return sorted(package for package in packages if any(re.search(pattern, package, re.IGNORECASE) for pattern in patterns)) + + +def systemd_state(units): + systemctl = find_binary(["systemctl"]) + if not systemctl: + return None + for unit in units: + output, error = run([systemctl, "show", unit, "--no-page", "--property=LoadState", "--property=ActiveState", "--property=SubState"]) + if error or not output: + continue + values = dict(line.split("=", 1) for line in output.splitlines() if "=" in line) + if values.get("LoadState") != "not-found": + return {"unit": unit, "load_state": values.get("LoadState"), "active_state": values.get("ActiveState"), "sub_state": values.get("SubState")} + return None + + +def parse_listeners(): + ss = find_binary(["ss"]) + if not ss: + raise RuntimeError("ss is not installed") + output, error = run([ss, "-H", "-ltnup", "( sport = :80 or sport = :443 )"]) + if error and output is None: + raise RuntimeError(error) + listeners = [] + missing_process_details = False + for line in (output or "").splitlines(): + fields = line.split() + if len(fields) < 5: + continue + protocol = fields[0].lower() + local_address = fields[4] + try: + port = int(local_address.rsplit(":", 1)[1]) + except (IndexError, ValueError): + continue + if port not in (80, 443): + continue + peer_address = fields[5] if len(fields) > 5 else None + process_field = " ".join(fields[6:]) + processes = [] + for name, pid, descriptor in re.findall(r'\("([^"\\]*(?:\\.[^"\\]*)*)",pid=(\d+),fd=(\d+)\)', process_field): + process_name = bytes(name, "utf-8").decode("unicode_escape") + processes.append({"name": process_name, "pid": int(pid), "fd": int(descriptor), "command": process_command(pid)}) + if not processes: + missing_process_details = True + listeners.append({ + "protocol": protocol, + "state": fields[1].upper(), + "local_address": local_address, + "peer_address": peer_address, + "port": port, + "processes": processes, + }) + return listeners, missing_process_details + + +def process_command(pid): + try: + raw = pathlib.Path(f"/proc/{pid}/cmdline").read_bytes() + except OSError: + return None + return " ".join(part.decode(errors="replace") for part in raw.split(b"\0") if part) or None + + +def discover_solutions(packages): + solutions = [] + for definition in SERVER_DEFINITIONS: + binary = find_binary(definition["binaries"]) + matching_packages = package_matches(packages, definition["packages"]) + unit = systemd_state(definition["units"]) + installed = bool(binary or matching_packages or unit) + state = "Not installed" + if installed: + state = "Installed" + if unit: + state = f"Service {unit['active_state'] or 'unknown'}" + solutions.append({ + "name": definition["name"], + "installed": installed, + "binary": binary, + "packages": matching_packages, + "service": unit, + "status": state, + }) + return solutions + + +def site_symlinks(directory): + root = pathlib.Path(directory) + links = [] + if not root.is_dir(): + return links + for path in sorted(root.iterdir()): + if path.is_symlink(): + try: + target = str(path.resolve(strict=True)) + except OSError: + target = os.readlink(path) + links.append({"path": str(path), "target": target}) + return links + + +def directive_values(block, directive): + values = [] + for line in block.splitlines(): + line = line.split("#", 1)[0].strip() + match = re.match(rf"{re.escape(directive)}\s+(.+?)\s*;?\s*$", line, re.IGNORECASE) + if match: + values.extend(match.group(1).strip('"\'').split()) + return values + + +def urls_for_hosts(hosts, ports): + urls = [] + for host in hosts: + if host == "_" or host.startswith("~") or "*" in host: + continue + for port in sorted(set(ports)): + scheme = "https" if port == 443 else "http" + suffix = "" if (scheme == "http" and port == 80) or (scheme == "https" and port == 443) else f":{port}" + urls.append(f"{scheme}://{host}{suffix}") + return urls + + +def apache_vhosts(): + control = find_binary(["apache2ctl", "apachectl", "httpd"]) + output = None + control_error = None + if control: + args = ["-S"] if pathlib.Path(control).name != "httpd" else ["-S"] + output, control_error = run([control, *args]) + links = site_symlinks("/etc/apache2/sites-enabled") + sources = [] + for link in links: + target = pathlib.Path(link["target"]) + if target.is_file(): + sources.append((target, link["path"])) + if not sources: + for config_dir in ("/etc/httpd/conf.d", "/etc/apache2/sites-enabled"): + root = pathlib.Path(config_dir) + if root.is_dir(): + sources.extend((path.resolve(), str(path) if path.is_symlink() else None) for path in sorted(root.glob("*.conf")) if path.is_file()) + vhosts = [] + seen = set() + for source, enabled_link in sources: + try: + text = source.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + for address, block in re.findall(r"]+)>(.*?)", text, re.IGNORECASE | re.DOTALL): + key = (str(source), address, block) + if key in seen: + continue + seen.add(key) + hosts = directive_values(block, "ServerName") + directive_values(block, "ServerAlias") + ports = [int(value) for value in re.findall(r":(80|443)\b", address)] + if not ports: + ports = [port for port in (80, 443) if re.search(rf"^\s*Listen\s+(?:[^\s:]+:)?{port}(?:\s|$)", text, re.IGNORECASE | re.MULTILINE)] + root_values = directive_values(block, "DocumentRoot") + vhosts.append({ + "server": "Apache HTTP Server", + "hosts": hosts, + "listen": address.strip(), + "ports": ports, + "document_root": root_values[0] if root_values else None, + "config_file": str(source), + "enabled_link": enabled_link, + "urls": urls_for_hosts(hosts, ports), + }) + if not vhosts and output: + for line in output.splitlines(): + match = re.match(r"\s*(\S+)\s+(.+?)\s+\((.+):(\d+)\)\s*$", line) + if not match: + continue + address, host, config, _ = match.groups() + if not config.startswith("/") or not ("80" in address or "443" in address): + continue + host = re.sub(r"^(?:default server|namevhost)\s+", "", host) + hosts = [] if host in ("_default_", "*", "default server") else [host] + ports = [port for port in (80, 443) if re.search(rf":{port}\b", address)] + vhosts.append({"server": "Apache HTTP Server", "hosts": hosts, "listen": address, "ports": ports, "document_root": None, "config_file": config, "enabled_link": None, "urls": urls_for_hosts(hosts, ports)}) + return vhosts, links, control_error + + +def nginx_vhosts(): + executable = find_binary(["nginx"]) + if not executable: + return [], [], None + output, error = run([executable, "-T"], timeout=20) + links = site_symlinks("/etc/nginx/sites-enabled") + link_by_target = {} + for link in links: + link_by_target[os.path.realpath(link["target"])] = link["path"] + if not output: + return [], links, error + vhosts = [] + current_file = None + block_lines = [] + block_depth = 0 + in_server = False + for line in output.splitlines(): + file_marker = re.match(r"# configuration file (.+):\s*$", line) + if file_marker: + current_file = file_marker.group(1) + clean = line.split("#", 1)[0] + if not in_server and re.search(r"\bserver\s*\{", clean): + in_server = True + block_lines = [clean] + block_depth = clean.count("{") - clean.count("}") + elif in_server: + block_lines.append(clean) + block_depth += clean.count("{") - clean.count("}") + if in_server and block_depth <= 0: + block = "\n".join(block_lines) + hosts = directive_values(block, "server_name") + listens = directive_values(block, "listen") + roots = directive_values(block, "root") + ports = [] + for listen in listens: + port_match = re.search(r"(?:^|\s|:)(80|443)(?:\s|$)", listen) + if port_match: + ports.append(int(port_match.group(1))) + elif not re.search(r"\d", listen): + ports.append(80) + if hosts or listens: + vhosts.append({ + "server": "Nginx", + "hosts": hosts, + "listen": ", ".join(listens), + "ports": sorted(set(ports)), + "document_root": roots[0] if roots else None, + "config_file": current_file, + "enabled_link": link_by_target.get(os.path.realpath(current_file)) if current_file else None, + "urls": urls_for_hosts(hosts, ports), + }) + in_server = False + block_lines = [] + block_depth = 0 + return vhosts, links, error + + +def collect_report(): + packages = installed_packages() + solutions = discover_solutions(packages) + listeners, missing_process_details = parse_listeners() + apache, apache_links, apache_error = apache_vhosts() + nginx, nginx_links, nginx_error = nginx_vhosts() + warnings = [] + if missing_process_details: + warnings.append("Some listeners do not show process/PID details. Run with sudo to identify services owned by other users.") + if apache_error and any(item["name"] == "Apache HTTP Server" and item["installed"] for item in solutions): + warnings.append(f"Apache vhost query: {apache_error}") + if nginx_error and any(item["name"] == "Nginx" and item["installed"] for item in solutions): + warnings.append(f"Nginx config query: {nginx_error}") + for listener in listeners: + if not listener["processes"]: + listener["service_guess"] = None + continue + names = {process["name"].lower() for process in listener["processes"]} + matches = [] + for solution in solutions: + names_to_match = {pathlib.Path(solution["binary"]).name.lower()} if solution["binary"] else set() + names_to_match.add(solution["name"].lower().replace(" ", "")) + if names.intersection(names_to_match) or any(name.startswith(("apache2", "httpd")) and solution["name"].startswith("Apache") for name in names): + matches.append(solution["name"]) + listener["service_guess"] = matches + return { + "generated_at": datetime.datetime.now().astimezone().isoformat(timespec="seconds"), + "listeners": listeners, + "solutions": solutions, + "virtual_hosts": apache + nginx, + "enabled_site_links": apache_links + nginx_links, + "warnings": warnings, + } + + +def render_text(report): + lines = ["WEB SERVICE STATUS REPORT", "=========================", f"Generated: {report['generated_at']}", "", "INSTALLED WEB SERVERS / REVERSE PROXIES", "---------------------------------------"] + installed = [solution for solution in report["solutions"] if solution["installed"]] + if installed: + for solution in installed: + service = solution.get("service") + service_text = f"; {service['unit']} {service['active_state']} ({service['sub_state']})" if service else "" + lines.append(f"{solution['name']}: installed{service_text}") + if solution["binary"]: + lines.append(f" Binary: {solution['binary']}") + if solution["packages"]: + lines.append(f" Packages: {', '.join(solution['packages'])}") + else: + lines.append("No known web server or reverse proxy installation detected.") + + lines.extend(["", "LISTENERS ON TCP/UDP PORTS 80 AND 443", "------------------------------------"]) + if report["listeners"]: + for listener in report["listeners"]: + process_text = ", ".join( + f"{process['name']} (PID {process['pid']})" + (f": {process['command']}" if process.get("command") else "") + for process in listener["processes"] + ) or "process details unavailable" + guess = f"; detected service: {', '.join(listener['service_guess'])}" if listener.get("service_guess") else "" + lines.append(f"{listener['protocol'].upper()} {listener['local_address']} [{listener['state']}] — {process_text}{guess}") + else: + lines.append("No listening sockets found on ports 80 or 443.") + + lines.extend(["", "APACHE / NGINX VIRTUAL HOSTS", "----------------------------"]) + if report["virtual_hosts"]: + for vhost in report["virtual_hosts"]: + hosts = ", ".join(vhost["hosts"]) or "default / unnamed host" + lines.append(f"{vhost['server']}: {hosts}") + if vhost.get("listen"): + lines.append(f" Listen: {vhost['listen']}") + if vhost.get("document_root"): + lines.append(f" Document root: {vhost['document_root']}") + if vhost.get("config_file"): + lines.append(f" Config: {vhost['config_file']}") + if vhost.get("enabled_link"): + lines.append(f" Enabled link: {vhost['enabled_link']}") + for url in vhost["urls"]: + lines.append(f" URL: {url}") + else: + lines.append("No Apache/Nginx virtual hosts could be read, or neither server is installed.") + + lines.extend(["", "ENABLED SITE CONFIGURATION LINKS", "--------------------------------"]) + links = report["enabled_site_links"] + if links: + lines.extend(f"{link['path']} -> {link['target']}" for link in links) + else: + lines.append("No sites-enabled symlinks found.") + if report["warnings"]: + lines.extend(["", "NOTES", "-----"]) + lines.extend(f"- {warning}" for warning in report["warnings"]) + return "\n".join(lines) + "\n" + + +def render_html(report): + def esc(value): + return html.escape("Unavailable" if value is None or value == "" else str(value)) + + solution_rows = "".join( + f"{esc(item['name'])}{'Installed' if item['installed'] else 'Not detected'}{esc(item.get('service', {}).get('active_state') if item.get('service') else 'Unknown')}{esc(item.get('binary'))}{esc(', '.join(item['packages']) if item['packages'] else None)}" + for item in report["solutions"] + ) + listener_rows = [] + for item in report["listeners"]: + process_text = ", ".join("{} (PID {})".format(proc["name"], proc["pid"]) for proc in item["processes"]) or "Unavailable" + service_text = ", ".join(item.get("service_guess") or []) or "Unavailable" + listener_rows.append( + f"{esc(item['protocol'].upper())}{esc(item['local_address'])}" + f"{esc(process_text)}{esc(service_text)}" + ) + listener_rows = "".join(listener_rows) or "No listeners on ports 80/443 found." + host_rows = "" + for vhost in report["virtual_hosts"]: + urls = "
".join(f"{esc(url)}" for url in vhost["urls"]) + host_rows += f"{esc(vhost['server'])}{esc(', '.join(vhost['hosts']) or 'default / unnamed')}{esc(vhost['listen'])}{esc(vhost['document_root'])}{esc(vhost['config_file'])}{esc(vhost['enabled_link'])}{urls or '—'}" + host_rows = host_rows or "No virtual hosts could be read." + link_rows = "".join(f"{esc(item['path'])}{esc(item['target'])}" for item in report["enabled_site_links"]) + link_rows = link_rows or "No enabled-site symlinks found." + warning_html = "".join(f"
  • {esc(item)}
    • " for item in report["warnings"]) + return ( + "Web Services Report" + "
    " + f"

    Web Server and Reverse Proxy Report

    Generated {esc(report['generated_at'])}

    " + "

    Installed solutions

    " + solution_rows + "
    SolutionDetectedService stateBinaryPackages
    " + "

    Listeners on ports 80 / 443

    " + listener_rows + "
    ProtocolAddressProcessDetected service
    " + "

    Apache / Nginx virtual hosts

    " + host_rows + "
    ServerHostnamesListenDocument rootConfig fileEnabled linkURLs
    " + "

    Enabled site links

    " + link_rows + "
    LinkTarget
    " + + (f"

    Notes

      {warning_html}
    " if warning_html else "") + + "
    \n" + ) + + +def ask(prompt, default): + try: + answer = input(f"{prompt} [{default}]: ").strip() + except EOFError: + return default + return answer or default + + +def ask_yes_no(prompt, default): + marker = "Y/n" if default else "y/N" + try: + answer = input(f"{prompt} [{marker}]: ").strip().lower() + except EOFError: + return default + if not answer: + return default + return answer in ("y", "yes") + + +def infer_format(path, requested): + if requested: + return requested + suffix = pathlib.Path(path).suffix.lower().lstrip(".") + return suffix if suffix in ("txt", "json", "html") else "txt" + + +def render_report(report, format_name): + if format_name == "json": + return json.dumps(report, ensure_ascii=False, indent=2) + "\n" + if format_name == "html": + return render_html(report) + return render_text(report) + + +def save_report(path_text, format_name, force, interactive, report): + path = pathlib.Path(path_text).expanduser() + if not path.parent.exists() or not path.parent.is_dir(): + raise OSError(f"output directory does not exist: {path.parent}") + if path.exists() and not force: + if not interactive or not ask_yes_no(f"{path} already exists. Overwrite it?", False): + return False + rendered = render_report(report, format_name) + temporary = None + try: + with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as output: + temporary = pathlib.Path(output.name) + output.write(rendered) + os.replace(temporary, path) + finally: + if temporary is not None and temporary.exists(): + temporary.unlink() + print(f"Report saved: {path}", file=sys.stderr) + return True + + +def main(): + output_path, requested_format, force_value = sys.argv[1:4] + force = force_value == "1" + interactive = sys.stdin.isatty() and sys.stdout.isatty() + report = collect_report() + if output_path: + format_name = infer_format(output_path, requested_format) + if not save_report(output_path, format_name, force, interactive, report): + print("Report not written; existing file was not overwritten.", file=sys.stderr) + return 1 if not interactive else 0 + return 0 + if not interactive: + print(render_report(report, requested_format or "txt"), end="") + return 0 + print(render_text(report), end="") + if not ask_yes_no("Save this report to a file?", True): + return 0 + format_name = requested_format or ask("Report format (txt/json/html)", "txt").lower() + if format_name not in ("txt", "json", "html"): + print("Unsupported format; using txt.", file=sys.stderr) + format_name = "txt" + path = ask("Output path", str(pathlib.Path.home() / f"web-services-report.{format_name}")) + if not save_report(path, format_name, force, interactive, report): + print("Report not written; existing file was not overwritten.", file=sys.stderr) + return 0 + + +try: + raise SystemExit(main()) +except (OSError, RuntimeError, json.JSONDecodeError) as error: + print(f"Error: {error}", file=sys.stderr) + raise SystemExit(1) +PYCODE +)" "$OUTPUT_PATH" "$OUTPUT_FORMAT" "$FORCE" \ No newline at end of file