mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
5137465c6f
3529 invalid comments in 528 files, grandfathered in while the linter only saw the working-tree diff. Removal ran through the linter's own collectors, so detection and deletion cannot drift: a comment whose opener line is blank before it and whose closer line is blank after it loses all its lines, anything else loses only the comment region, and a line left holding a bare delimiter goes with it. Five files keep their comments behind a nocheck marker instead. Four are upstream configs whose inline docs are upstream's own (mautrix, systemd journald, nextcloud-oidc-login, MediaWiki). In littlejs the commented-out entries ARE the deactivation mechanism for an example tile, so deleting them would delete configuration. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
206 lines
7.4 KiB
Python
206 lines
7.4 KiB
Python
"""Resolve nginx path configuration and an optional domain-specific server
|
|
config path under ``<servers_dir>/<protocol>/<domain>.conf``.
|
|
|
|
The ``want=`` kwarg is intentionally ignored: want-path is ALWAYS the
|
|
first positional argument so callers can't accidentally pass it both
|
|
ways.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from ansible.errors import AnsibleError
|
|
from ansible.plugins.loader import lookup_loader
|
|
from ansible.plugins.lookup import LookupBase
|
|
|
|
# nocheck: lookup-cache-import
|
|
from utils.cache.applications import get_canonical_volumes
|
|
from utils.roles.applications.config import get
|
|
from utils.tls_common import as_str, want_get
|
|
|
|
|
|
def _volume_path_by_name(app_id: str, vol_name: str) -> str:
|
|
canonical = get_canonical_volumes(app_id)
|
|
entry = canonical.get(vol_name)
|
|
return as_str(entry.get("path", "")) if entry else ""
|
|
|
|
|
|
def _join(*parts: Any) -> str:
|
|
cleaned = [str(p).strip() for p in parts if str(p).strip()]
|
|
return str(Path(*cleaned)) if cleaned else ""
|
|
|
|
|
|
def _ensure_trailing_slash(p: str) -> str:
|
|
p = p.strip()
|
|
return p if not p or p.endswith("/") else p + "/"
|
|
|
|
|
|
def _normalize_protocol(value: str) -> str:
|
|
v = as_str(value).strip().lower()
|
|
if v in ("http", "https"):
|
|
return v
|
|
raise AnsibleError(
|
|
f"nginx: invalid protocol override '{value}' (expected http|https)"
|
|
)
|
|
|
|
|
|
def _dir_spec(path: str, mode: str) -> dict[str, str]:
|
|
path = as_str(path).strip()
|
|
mode = as_str(mode).strip()
|
|
if not path:
|
|
raise AnsibleError("nginx: empty path in directories.ensure")
|
|
if mode not in ("0700", "0755"):
|
|
raise AnsibleError(
|
|
f"nginx: invalid mode '{mode}' in directories.ensure (expected 0700|0755)"
|
|
)
|
|
return {"path": path, "mode": mode}
|
|
|
|
|
|
def _resolve_protocol_via_tls(
|
|
*, domain: str, variables: dict, loader: Any, templar: Any
|
|
) -> str:
|
|
try:
|
|
tls_lookup = lookup_loader.get("tls", loader=loader, templar=templar)
|
|
except Exception as exc:
|
|
raise AnsibleError(f"nginx: failed to load tls lookup: {exc}") from exc
|
|
|
|
protocol = tls_lookup.run([domain, "protocols.web"], variables=variables)[0]
|
|
|
|
protocol_s = as_str(protocol).strip().lower()
|
|
if protocol_s not in ("http", "https"):
|
|
raise AnsibleError(
|
|
f"nginx: unexpected protocol '{protocol_s}' for domain '{domain}'"
|
|
)
|
|
return protocol_s
|
|
|
|
|
|
class LookupModule(LookupBase):
|
|
def run(self, terms, variables: dict | None = None, **kwargs):
|
|
variables = variables or {}
|
|
terms = terms or []
|
|
|
|
if len(terms) not in (1, 2):
|
|
raise AnsibleError("nginx: requires want_path [, domain]")
|
|
|
|
want = as_str(terms[0]).strip()
|
|
if not want:
|
|
raise AnsibleError("nginx: want_path is empty")
|
|
|
|
domain = as_str(terms[1]).strip() if len(terms) == 2 else ""
|
|
|
|
applications = lookup_loader.get(
|
|
"applications", loader=self._loader, templar=getattr(self, "_templar", None)
|
|
).run([], variables=variables)[0]
|
|
|
|
proxy_app_id = as_str(kwargs.get("proxy_app_id", "svc-prx-openresty")).strip()
|
|
if not proxy_app_id:
|
|
raise AnsibleError("nginx: proxy_app_id is empty")
|
|
|
|
www_dir = _volume_path_by_name(proxy_app_id, "www")
|
|
nginx_dir = _volume_path_by_name(proxy_app_id, "nginx")
|
|
if not www_dir:
|
|
www_dir = get(applications, proxy_app_id, "volumes.www.path", strict=True)
|
|
if not nginx_dir:
|
|
nginx_dir = get(
|
|
applications, proxy_app_id, "volumes.nginx.path", strict=True
|
|
)
|
|
|
|
www_dir = _ensure_trailing_slash(as_str(www_dir))
|
|
nginx_dir = _ensure_trailing_slash(as_str(nginx_dir))
|
|
|
|
conf_dir = _ensure_trailing_slash(_join(nginx_dir, "conf.d"))
|
|
global_dir = _ensure_trailing_slash(_join(conf_dir, "global"))
|
|
servers_dir = _ensure_trailing_slash(_join(conf_dir, "servers"))
|
|
servers_http_dir = _ensure_trailing_slash(_join(servers_dir, "http"))
|
|
servers_https_dir = _ensure_trailing_slash(_join(servers_dir, "https"))
|
|
maps_dir = _ensure_trailing_slash(_join(conf_dir, "maps"))
|
|
streams_dir = _ensure_trailing_slash(_join(conf_dir, "streams"))
|
|
lua_dir = _ensure_trailing_slash(_join(conf_dir, "lua"))
|
|
|
|
data_html_dir = _ensure_trailing_slash(_join(www_dir, "public_html"))
|
|
data_files_dir = _ensure_trailing_slash(_join(www_dir, "public_files"))
|
|
data_cdn_dir = _ensure_trailing_slash(_join(www_dir, "public_cdn"))
|
|
data_global_dir = _ensure_trailing_slash(_join(www_dir, "global"))
|
|
|
|
cache_general_dir = "/tmp/cache_nginx_general/" # noqa: S108
|
|
cache_image_dir = "/tmp/cache_nginx_image/" # noqa: S108
|
|
|
|
ensure: list[dict[str, str]] = [
|
|
_dir_spec(nginx_dir, "0755"),
|
|
_dir_spec(conf_dir, "0755"),
|
|
_dir_spec(global_dir, "0755"),
|
|
_dir_spec(servers_dir, "0755"),
|
|
_dir_spec(servers_http_dir, "0755"),
|
|
_dir_spec(servers_https_dir, "0755"),
|
|
_dir_spec(maps_dir, "0755"),
|
|
_dir_spec(streams_dir, "0755"),
|
|
_dir_spec(lua_dir, "0755"),
|
|
_dir_spec(www_dir, "0755"),
|
|
_dir_spec(data_html_dir, "0755"),
|
|
_dir_spec(data_files_dir, "0755"),
|
|
_dir_spec(data_cdn_dir, "0755"),
|
|
_dir_spec(data_global_dir, "0755"),
|
|
_dir_spec(cache_general_dir, "0700"),
|
|
_dir_spec(cache_image_dir, "0700"),
|
|
]
|
|
|
|
resolved: dict[str, Any] = {
|
|
"files": {
|
|
"configuration": _join(nginx_dir, "nginx.conf"),
|
|
},
|
|
"directories": {
|
|
"configuration": {
|
|
"base": conf_dir,
|
|
"global": global_dir,
|
|
"servers": servers_dir,
|
|
"maps": maps_dir,
|
|
"streams": streams_dir,
|
|
"lua": lua_dir,
|
|
"http_includes": [
|
|
global_dir,
|
|
maps_dir,
|
|
servers_http_dir,
|
|
servers_https_dir,
|
|
],
|
|
},
|
|
"data": {
|
|
"www": www_dir,
|
|
"well_known": "/usr/share/nginx/well-known/",
|
|
"html": data_html_dir,
|
|
"files": data_files_dir,
|
|
"cdn": data_cdn_dir,
|
|
"global": data_global_dir,
|
|
},
|
|
"cache": {
|
|
"general": cache_general_dir,
|
|
"image": cache_image_dir,
|
|
},
|
|
"ensure": ensure,
|
|
"ensure_paths": [d["path"] for d in ensure],
|
|
},
|
|
}
|
|
|
|
if domain:
|
|
protocol_override = kwargs.get("protocol")
|
|
protocol = (
|
|
_resolve_protocol_via_tls(
|
|
domain=domain,
|
|
variables=variables,
|
|
loader=getattr(self, "_loader", None),
|
|
templar=getattr(self, "_templar", None),
|
|
)
|
|
if not protocol_override
|
|
else _normalize_protocol(protocol_override)
|
|
)
|
|
|
|
resolved["domain"] = {
|
|
"name": domain,
|
|
"protocol": protocol,
|
|
"protocol_overridden": bool(protocol_override),
|
|
}
|
|
resolved["files"]["domain"] = _join(servers_dir, protocol, f"{domain}.conf")
|
|
|
|
return [want_get(resolved, want)]
|