Evelyn Chen
5c2630b972
Test / contract (pull_request) Successful in 6s
Authored-By: OpenAI (GPT-5) <noreply@openai.com>
531 lines
24 KiB
Python
531 lines
24 KiB
Python
"""Shared utilities for the site-publish action."""
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path, PurePosixPath
|
|
from urllib.parse import urlparse
|
|
|
|
import yaml
|
|
from jinja2 import Environment, FileSystemLoader, StrictUndefined
|
|
|
|
APPS_REPO = "fritzlab/apps"
|
|
GITEA_HOST = "code.fritzlab.net"
|
|
NAMESPACE = "websites"
|
|
DEFAULT_S3_ENDPOINT = "http://garage-s3.storage.svc:3900"
|
|
DEFAULT_WEBSITE_SUFFIX = "web.sjc001.fritzlab.net"
|
|
DEFAULT_CACHE_CONTROL = "public, max-age=0, must-revalidate"
|
|
|
|
EXCLUDE_FILES = {
|
|
".git", ".gitea", ".gitignore", "site.yaml", "build", ".site-publish",
|
|
"Makefile", "README.md", "CLAUDE.md", "Dockerfile", ".dockerignore",
|
|
"go.mod", "go.sum",
|
|
}
|
|
VALID_TYPES = {"static", "hugo", "mkdocs"}
|
|
NAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
|
|
ENV_RE = re.compile(r"^[A-Z_][A-Z0-9_]*$")
|
|
ACCESS_KEY_ENV_RE = re.compile(r"^([A-Z][A-Z0-9_]*)_S3_ACCESS_KEY$")
|
|
BUCKET_RE = re.compile(r"^[a-z0-9](?:[a-z0-9.-]{1,61}[a-z0-9])?$")
|
|
MIDDLEWARE_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9_-]{0,61}[A-Za-z0-9])?$")
|
|
|
|
DOCKER_DEPRECATION_MSG = """\
|
|
type: docker is no longer supported by action/site-publish.
|
|
|
|
site-publish handles only static-content sites (static, hugo, mkdocs)
|
|
that ship to Garage S3. For containerized web apps, use the standard
|
|
image-producer chain:
|
|
|
|
- uses: action/image-build@v1 # build + smoke-test
|
|
- uses: action/image-push@v1 # push + prune
|
|
- uses: action/image-deploy@v1 # apps repo image-pin
|
|
|
|
Hand-author your apps-repo manifests once (Deployment, Service, Ingress,
|
|
Certificate, kustomization with images: block) under
|
|
sjc001/websites/<repo>/manifests/. image-deploy will pin the tag on
|
|
every CI run. See action/image-deploy README and
|
|
sjc001/websites/rainsounds.vino.network/manifests/ for the canonical
|
|
example.\
|
|
"""
|
|
|
|
|
|
class ConfigError(ValueError):
|
|
"""A site.yaml contract violation."""
|
|
|
|
|
|
def k8s_name(name):
|
|
return name.replace(".", "-")
|
|
|
|
|
|
def env(key, default=None):
|
|
val = os.environ.get(key, default)
|
|
if val is None or val == "":
|
|
die(f"Missing required env var: {key}")
|
|
return val
|
|
|
|
|
|
def die(msg):
|
|
print(f"ERROR: {msg}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def run(cmd, *, display=None, **kwargs):
|
|
"""Run an argv command, printing only a safe display form."""
|
|
if not isinstance(cmd, (list, tuple)):
|
|
raise TypeError("run() requires an argv list")
|
|
shown = display if display is not None else " ".join(str(part) for part in cmd)
|
|
print(f" $ {shown}")
|
|
return subprocess.run(cmd, check=True, **kwargs)
|
|
|
|
|
|
def _mapping(value, label):
|
|
if not isinstance(value, dict):
|
|
raise ConfigError(f"{label} must be a mapping")
|
|
return value
|
|
|
|
|
|
def _list(value, label):
|
|
if not isinstance(value, list):
|
|
raise ConfigError(f"{label} must be a list")
|
|
return value
|
|
|
|
|
|
def _known_keys(value, allowed, label):
|
|
unknown = set(value) - set(allowed)
|
|
if unknown:
|
|
raise ConfigError(f"{label} has unknown fields: {', '.join(sorted(unknown))}")
|
|
|
|
|
|
def _strings(value, label):
|
|
values = _list(value or [], label)
|
|
if any(not isinstance(item, str) or not item for item in values):
|
|
raise ConfigError(f"{label} must be a list of non-empty strings")
|
|
return values
|
|
|
|
|
|
def _hostname(value, label):
|
|
if not isinstance(value, str) or len(value) > 253 or value.endswith("."):
|
|
raise ConfigError(f"{label} must be a lowercase DNS hostname without a trailing dot")
|
|
labels = value.split(".")
|
|
if len(labels) < 2 or any(not NAME_RE.fullmatch(part) for part in labels):
|
|
raise ConfigError(f"{label} must be a lowercase DNS hostname without a trailing dot")
|
|
return value
|
|
|
|
|
|
def _aliases(value, domain):
|
|
aliases = _strings(value or [], "aliases")
|
|
aliases = [_hostname(alias, "aliases entry") for alias in aliases]
|
|
if len(aliases) != len(set(aliases)) or domain in aliases:
|
|
raise ConfigError("aliases must be unique and cannot repeat domain")
|
|
return aliases
|
|
|
|
|
|
def _relative_path(value, label, *, allow_root=False):
|
|
value = "" if value is None else value
|
|
if not isinstance(value, str):
|
|
raise ConfigError(f"{label} must be a string")
|
|
if value == "/" and allow_root:
|
|
return ""
|
|
path = PurePosixPath(value)
|
|
if path.is_absolute() or ".." in path.parts:
|
|
raise ConfigError(f"{label} must be a relative path without '..'")
|
|
normalized = str(path).strip("/")
|
|
return "" if normalized == "." else normalized
|
|
|
|
|
|
def _route_path(value, label):
|
|
if not isinstance(value, str) or not value.startswith("/"):
|
|
raise ConfigError(f"{label} must start with '/'")
|
|
if "//" in value or "?" in value or "#" in value or ".." in value.split("/"):
|
|
raise ConfigError(f"{label} is not a canonical URL path")
|
|
normalized = value.rstrip("/") or "/"
|
|
if not re.fullmatch(r"/[A-Za-z0-9._~/-]*", normalized):
|
|
raise ConfigError(f"{label} contains unsupported URL path characters")
|
|
return normalized
|
|
|
|
|
|
def _middlewares(value, label):
|
|
names = _strings(value, label)
|
|
if any(not MIDDLEWARE_RE.fullmatch(name) for name in names):
|
|
raise ConfigError(f"{label} contains an invalid middleware name")
|
|
if len(names) != len(set(names)):
|
|
raise ConfigError(f"{label} contains duplicate middleware names")
|
|
return names
|
|
|
|
|
|
def _cache_control(value, label):
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ConfigError(f"{label} must be a non-empty Cache-Control value")
|
|
directives = [part.strip().lower() for part in value.split(",")]
|
|
names = [part.split("=", 1)[0] for part in directives]
|
|
if len(names) != len(set(names)):
|
|
raise ConfigError(f"{label} repeats a Cache-Control directive")
|
|
present = set(names)
|
|
if {"public", "private"} <= present:
|
|
raise ConfigError(f"{label} cannot be both public and private")
|
|
ages = {}
|
|
for part in directives:
|
|
name = part.split("=", 1)[0]
|
|
if name in {"max-age", "s-maxage"}:
|
|
raw = part.split("=", 1)[1] if "=" in part else ""
|
|
if not raw.isdigit():
|
|
raise ConfigError(f"{label} {name} must be a non-negative integer")
|
|
ages[name] = int(raw)
|
|
max_age = ages.get("max-age")
|
|
if "immutable" in present and (not max_age or present & {"no-store", "no-cache", "must-revalidate"}):
|
|
raise ConfigError(f"{label} immutable requires positive max-age without revalidation")
|
|
if "no-store" in present and any(ages.values()):
|
|
raise ConfigError(f"{label} no-store contradicts positive caching")
|
|
if "private" in present and ages.get("s-maxage"):
|
|
raise ConfigError(f"{label} private contradicts shared-cache max-age")
|
|
return ", ".join(part.strip() for part in value.split(","))
|
|
|
|
|
|
def _site_type(value, label):
|
|
site_type = value or "static"
|
|
if site_type == "docker":
|
|
raise ConfigError(DOCKER_DEPRECATION_MSG)
|
|
if site_type not in VALID_TYPES:
|
|
raise ConfigError(f"Unknown {label}: {site_type} (valid: {', '.join(sorted(VALID_TYPES))})")
|
|
return site_type
|
|
|
|
|
|
def _endpoint(value, label):
|
|
parsed = urlparse(value)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.path not in {"", "/"}:
|
|
raise ConfigError(f"{label} must be an http(s) origin without a path")
|
|
return value.rstrip("/")
|
|
|
|
|
|
def _legacy_config(raw, site_name):
|
|
if not isinstance(raw.get("tidy", True), bool):
|
|
raise ConfigError("tidy must be a boolean")
|
|
if not isinstance(raw.get("enabled", True), bool):
|
|
raise ConfigError("enabled must be a boolean")
|
|
artifact = {
|
|
"name": "site",
|
|
"type": _site_type(raw.get("type", "static"), "site type"),
|
|
"content_dir": _relative_path(raw.get("content_dir", ""), "content_dir"),
|
|
"tidy": raw.get("tidy", True),
|
|
"excludes": _strings(raw.get("excludes") or [], "excludes"),
|
|
"build_dir": "build/html",
|
|
"bucket": site_name,
|
|
"s3_endpoint": os.environ.get("GARAGE_S3_ENDPOINT") or DEFAULT_S3_ENDPOINT,
|
|
"website_authority": "garage-s3.storage.svc.k8s.sjc001.fritzlab.net",
|
|
"credentials": {"access_key_env": "AWS_ACCESS_KEY_ID", "secret_key_env": "AWS_SECRET_ACCESS_KEY"},
|
|
"cache_rules": [{"path": "", "cache_control": DEFAULT_CACHE_CONTROL}],
|
|
}
|
|
return {
|
|
"version": 1,
|
|
"compatibility": "single-surface-v1",
|
|
"domain": raw["domain"],
|
|
"aliases": _aliases(raw.get("aliases"), raw["domain"]),
|
|
"enabled": raw.get("enabled", True),
|
|
"artifacts": [artifact],
|
|
"routes": [{
|
|
"name": "site", "path": "/", "artifact": "site", "access": "legacy",
|
|
"access_middleware": None,
|
|
"middlewares": _middlewares(raw.get("middlewares") or [], "middlewares"),
|
|
}],
|
|
}
|
|
|
|
|
|
def _artifact(item, index):
|
|
label = f"artifacts[{index}]"
|
|
item = _mapping(item, label)
|
|
_known_keys(item, {"name", "type", "content_dir", "tidy", "excludes", "publish", "cache"}, label)
|
|
name = item.get("name")
|
|
if not isinstance(name, str) or not NAME_RE.fullmatch(name):
|
|
raise ConfigError(f"{label}.name must be a DNS label")
|
|
publish = _mapping(item.get("publish"), f"{label}.publish")
|
|
_known_keys(publish, {"bucket", "credentials"}, f"{label}.publish")
|
|
bucket = publish.get("bucket")
|
|
if not isinstance(bucket, str) or not BUCKET_RE.fullmatch(bucket):
|
|
raise ConfigError(f"{label}.publish.bucket is not a valid bucket name")
|
|
credentials = _mapping(publish.get("credentials"), f"{label}.publish.credentials")
|
|
_known_keys(credentials, {"access_key_env", "secret_key_env"}, f"{label}.publish.credentials")
|
|
normalized_credentials = {}
|
|
for key in ("access_key_env", "secret_key_env"):
|
|
value = credentials.get(key)
|
|
if not isinstance(value, str) or not ENV_RE.fullmatch(value):
|
|
raise ConfigError(f"{label}.publish.credentials.{key} must name an environment variable")
|
|
normalized_credentials[key] = value
|
|
access_match = ACCESS_KEY_ENV_RE.fullmatch(normalized_credentials["access_key_env"])
|
|
expected_secret = (
|
|
f"{access_match.group(1)}_S3_SECRET_KEY" if access_match else None
|
|
)
|
|
if normalized_credentials["secret_key_env"] != expected_secret:
|
|
raise ConfigError(
|
|
f"{label}.publish.credentials must be a matched "
|
|
"<NAME>_S3_ACCESS_KEY and <NAME>_S3_SECRET_KEY pair"
|
|
)
|
|
cache = _mapping(item.get("cache"), f"{label}.cache")
|
|
_known_keys(cache, {"rules"}, f"{label}.cache")
|
|
rules = _list(cache.get("rules"), f"{label}.cache.rules")
|
|
if not rules:
|
|
raise ConfigError(f"{label}.cache.rules must declare a '/' default")
|
|
cache_rules, paths = [], set()
|
|
for rule_index, rule in enumerate(rules):
|
|
rule_label = f"{label}.cache.rules[{rule_index}]"
|
|
rule = _mapping(rule, rule_label)
|
|
_known_keys(rule, {"path", "cache_control"}, rule_label)
|
|
path = _relative_path(rule.get("path"), f"{rule_label}.path", allow_root=True)
|
|
if path in paths:
|
|
raise ConfigError(f"{label}.cache.rules has duplicate path /{path}")
|
|
paths.add(path)
|
|
cache_rules.append({"path": path, "cache_control": _cache_control(
|
|
rule.get("cache_control"), f"{rule_label}.cache_control"
|
|
)})
|
|
if "" not in paths:
|
|
raise ConfigError(f"{label}.cache.rules must declare a '/' default")
|
|
cache_rules.sort(key=lambda rule: (len(PurePosixPath(rule["path"]).parts), rule["path"]))
|
|
immutable_paths = [rule["path"] for rule in cache_rules
|
|
if "immutable" in {part.strip().lower().split("=", 1)[0]
|
|
for part in rule["cache_control"].split(",")}]
|
|
if "" in immutable_paths:
|
|
raise ConfigError(f"{label}.cache.rules immutable paths must be narrower than '/'")
|
|
for path in immutable_paths:
|
|
if any(other.startswith(f"{path}/") for other in paths):
|
|
raise ConfigError(f"{label}.cache.rules cannot nest another policy under immutable /{path}")
|
|
authority = f"{bucket}.{DEFAULT_WEBSITE_SUFFIX}"
|
|
if not isinstance(item.get("tidy", True), bool):
|
|
raise ConfigError(f"{label}.tidy must be a boolean")
|
|
return {
|
|
"name": name,
|
|
"type": _site_type(item.get("type", "static"), f"{label}.type"),
|
|
"content_dir": _relative_path(item.get("content_dir", ""), f"{label}.content_dir"),
|
|
"tidy": item.get("tidy", True),
|
|
"excludes": _strings(item.get("excludes") or [], f"{label}.excludes"),
|
|
"build_dir": f".site-publish/{name}/html",
|
|
"bucket": bucket,
|
|
"s3_endpoint": DEFAULT_S3_ENDPOINT,
|
|
"website_authority": authority,
|
|
"credentials": normalized_credentials,
|
|
"cache_rules": cache_rules,
|
|
}
|
|
|
|
|
|
def _route(item, index):
|
|
label = f"routes[{index}]"
|
|
item = _mapping(item, label)
|
|
_known_keys(item, {"name", "path", "artifact", "access", "middlewares"}, label)
|
|
name = item.get("name")
|
|
if not isinstance(name, str) or not NAME_RE.fullmatch(name):
|
|
raise ConfigError(f"{label}.name must be a DNS label")
|
|
access = _mapping(item.get("access"), f"{label}.access")
|
|
_known_keys(access, {"mode", "middleware"}, f"{label}.access")
|
|
mode, middleware = access.get("mode"), access.get("middleware")
|
|
if mode not in {"public", "protected"}:
|
|
raise ConfigError(f"{label}.access.mode must be public or protected")
|
|
if mode == "protected" and (not isinstance(middleware, str) or not MIDDLEWARE_RE.fullmatch(middleware)):
|
|
raise ConfigError(f"{label}.access.middleware is required for protected access")
|
|
if mode == "public" and middleware is not None:
|
|
raise ConfigError(f"{label}.access.middleware is forbidden for public access")
|
|
middlewares = _middlewares(item.get("middlewares") or [], f"{label}.middlewares")
|
|
if middleware in middlewares:
|
|
raise ConfigError(f"{label} repeats its access middleware")
|
|
artifact = item.get("artifact")
|
|
if not isinstance(artifact, str):
|
|
raise ConfigError(f"{label}.artifact must name an artifact")
|
|
return {
|
|
"name": name, "path": _route_path(item.get("path"), f"{label}.path"),
|
|
"artifact": artifact, "access": mode, "access_middleware": middleware,
|
|
"middlewares": middlewares,
|
|
}
|
|
|
|
|
|
def _validate_multi(cfg):
|
|
artifacts, routes = cfg["artifacts"], cfg["routes"]
|
|
artifact_names = [item["name"] for item in artifacts]
|
|
route_names = [item["name"] for item in routes]
|
|
route_paths = [item["path"] for item in routes]
|
|
if len(artifact_names) != len(set(artifact_names)):
|
|
raise ConfigError("artifact names must be unique")
|
|
if len(route_names) != len(set(route_names)):
|
|
raise ConfigError("route names must be unique")
|
|
if len(route_paths) != len(set(route_paths)):
|
|
raise ConfigError("route paths are ambiguous after normalization")
|
|
if "/" not in route_paths:
|
|
raise ConfigError("routes must declare a '/' catch-all")
|
|
artifact_by_name = {item["name"]: item for item in artifacts}
|
|
references = {name: [] for name in artifact_by_name}
|
|
for route in routes:
|
|
if route["artifact"] not in artifact_by_name:
|
|
raise ConfigError(f"route {route['name']} references unknown artifact {route['artifact']}")
|
|
references[route["artifact"]].append(route)
|
|
for name, used_by in references.items():
|
|
if len(used_by) != 1:
|
|
raise ConfigError(f"artifact {name} must be referenced by exactly one route")
|
|
buckets, authorities, credential_owners = {}, {}, {}
|
|
for route in routes:
|
|
artifact = artifact_by_name[route["artifact"]]
|
|
if artifact["bucket"] in buckets:
|
|
other_access, other_name = buckets[artifact["bucket"]]
|
|
if other_access != route["access"]:
|
|
raise ConfigError(f"bucket {artifact['bucket']} cannot be reused by protected and public routes")
|
|
raise ConfigError(f"bucket {artifact['bucket']} must belong to one artifact ({other_name})")
|
|
buckets[artifact["bucket"]] = (route["access"], artifact["name"])
|
|
if artifact["website_authority"] in authorities:
|
|
raise ConfigError(
|
|
f"website authority {artifact['website_authority']} must belong to one artifact"
|
|
)
|
|
authorities[artifact["website_authority"]] = artifact["name"]
|
|
for variable in artifact["credentials"].values():
|
|
if variable in credential_owners:
|
|
raise ConfigError(
|
|
f"publication credential {variable} is reused by artifacts "
|
|
f"{credential_owners[variable]} and {artifact['name']}"
|
|
)
|
|
credential_owners[variable] = artifact["name"]
|
|
for cache_rule in artifact["cache_rules"]:
|
|
directives = {part.strip().lower().split("=", 1)[0]
|
|
for part in cache_rule["cache_control"].split(",")}
|
|
if route["access"] == "protected" and "public" in directives:
|
|
raise ConfigError(f"protected route {route['name']} cannot use public cache policy")
|
|
if route["access"] == "protected" and not directives & {"private", "no-store"}:
|
|
raise ConfigError(f"protected route {route['name']} cache policy must be private or no-store")
|
|
if route["access"] == "protected" and "s-maxage" in directives:
|
|
raise ConfigError(f"protected route {route['name']} cannot use shared-cache max-age")
|
|
if route["access"] == "public" and "private" in directives:
|
|
raise ConfigError(f"public route {route['name']} cannot use private cache policy")
|
|
root = next(route for route in routes if route["path"] == "/")
|
|
if any(route["access"] == "protected" for route in routes) and root["access"] == "public":
|
|
raise ConfigError("a public '/' catch-all would expose unmatched protected content")
|
|
|
|
|
|
def normalize_site_config(raw, site_name):
|
|
raw = _mapping(raw, "site.yaml")
|
|
domain = _hostname(raw.get("domain"), "domain")
|
|
if not isinstance(raw.get("enabled", True), bool):
|
|
raise ConfigError("enabled must be a boolean")
|
|
if ("artifacts" in raw) != ("routes" in raw):
|
|
raise ConfigError("artifacts and routes must be declared together")
|
|
if "artifacts" not in raw:
|
|
return _legacy_config(raw, site_name)
|
|
legacy_fields = {"type", "content_dir", "tidy", "excludes", "middlewares"} & set(raw)
|
|
if legacy_fields:
|
|
raise ConfigError("legacy fields cannot be mixed with artifacts/routes: " + ", ".join(sorted(legacy_fields)))
|
|
_known_keys(raw, {"domain", "aliases", "enabled", "artifacts", "routes"}, "site.yaml")
|
|
artifacts = [_artifact(item, index) for index, item in enumerate(_list(raw["artifacts"], "artifacts"))]
|
|
routes = [_route(item, index) for index, item in enumerate(_list(raw["routes"], "routes"))]
|
|
if not artifacts or not routes:
|
|
raise ConfigError("artifacts and routes must not be empty")
|
|
cfg = {
|
|
"version": 2, "compatibility": None, "domain": domain,
|
|
"aliases": _aliases(raw.get("aliases"), domain),
|
|
"enabled": raw.get("enabled", True),
|
|
"artifacts": sorted(artifacts, key=lambda item: item["name"]),
|
|
"routes": sorted(routes, key=lambda item: (-len(item["path"]), item["path"], item["name"])),
|
|
}
|
|
_validate_multi(cfg)
|
|
for route in cfg["routes"]:
|
|
if len(f"{k8s_name(site_name)}-{route['name']}") > 63:
|
|
raise ConfigError(f"route {route['name']} makes the generated Service name exceed 63 characters")
|
|
return cfg
|
|
|
|
|
|
def validate_artifact_inputs(site_dir, cfg):
|
|
"""Reject source containment before a public or protected build starts."""
|
|
root = Path(site_dir).resolve()
|
|
sources = []
|
|
for artifact in cfg["artifacts"]:
|
|
source = (root / artifact["content_dir"]).resolve()
|
|
if source != root and root not in source.parents:
|
|
raise ConfigError(
|
|
f"artifact {artifact['name']} content_dir resolves outside the repository"
|
|
)
|
|
sources.append((artifact["name"], source))
|
|
for index, (name, source) in enumerate(sources):
|
|
for other_name, other_source in sources[index + 1:]:
|
|
if source == other_source or source in other_source.parents or other_source in source.parents:
|
|
raise ConfigError(
|
|
f"artifact build inputs overlap after resolution: {name} and {other_name}"
|
|
)
|
|
|
|
|
|
def parse_site_yaml(site_dir, site_name=None):
|
|
path = Path(site_dir) / "site.yaml"
|
|
if not path.exists():
|
|
die("site.yaml not found in repo root")
|
|
if site_name is None:
|
|
repo = os.environ.get("SITE_REPO", "")
|
|
site_name = repo.split("/", 1)[-1] if "/" in repo else Path(site_dir).name
|
|
try:
|
|
with open(path) as stream:
|
|
cfg = normalize_site_config(yaml.safe_load(stream), site_name)
|
|
except (ConfigError, yaml.YAMLError) as exc:
|
|
die(str(exc))
|
|
print("Site config:")
|
|
print(f" domain: {cfg['domain']}")
|
|
print(f" contract: {cfg['compatibility'] or 'split-surface-v2'}")
|
|
print(f" artifacts: {[item['name'] for item in cfg['artifacts']]}")
|
|
print(f" routes: {[(item['path'], item['access']) for item in cfg['routes']]}")
|
|
return cfg
|
|
|
|
|
|
def clone_apps(token):
|
|
"""Clone Apps without placing the credential in argv or output."""
|
|
user = env("CI_BOT_USER", "ci-bot")
|
|
apps_dir = Path("/tmp/apps-deploy")
|
|
if apps_dir.exists():
|
|
shutil.rmtree(apps_dir)
|
|
clone_env = git_auth_env(token)
|
|
url = f"https://{user}@{GITEA_HOST}/{APPS_REPO}.git"
|
|
run(["git", "clone", "--depth", "1", url, str(apps_dir)],
|
|
display=f"git clone --depth 1 https://{user}@{GITEA_HOST}/{APPS_REPO}.git {apps_dir}", env=clone_env)
|
|
run(["git", "-C", str(apps_dir), "config", "user.name", user])
|
|
run(["git", "-C", str(apps_dir), "config", "user.email", f"{user}@fritzlab.net"])
|
|
return apps_dir
|
|
|
|
|
|
def render_templates(action_dir, template_vars, app_dir, manifests_dir):
|
|
"""Render one certificate and deterministic per-route resources."""
|
|
jinja_env = Environment(loader=FileSystemLoader(str(Path(action_dir) / "templates")),
|
|
keep_trailing_newline=True, undefined=StrictUndefined)
|
|
manifests_dir.mkdir(parents=True, exist_ok=True)
|
|
for child in manifests_dir.iterdir():
|
|
if child.is_file() and child.suffix in {".yaml", ".yml"}:
|
|
child.unlink()
|
|
route_files = []
|
|
for route in template_vars["routes"]:
|
|
stem = "" if template_vars["compatibility"] else f"-{route['name']}"
|
|
for kind in ("service", "ingress"):
|
|
out_name = f"{kind}{stem}.yaml"
|
|
destination = manifests_dir / out_name
|
|
destination.write_text(jinja_env.get_template(f"{kind}.yaml.j2").render(
|
|
**template_vars, route=route
|
|
))
|
|
route_files.append(out_name)
|
|
print(f" Rendered {kind}.yaml.j2 -> {destination}")
|
|
common_vars = {**template_vars, "route_files": route_files}
|
|
for out_name, destination in {
|
|
"certificate.yaml": manifests_dir / "certificate.yaml",
|
|
"kustomization.yaml": manifests_dir / "kustomization.yaml",
|
|
"app.yaml": app_dir / "app.yaml",
|
|
}.items():
|
|
destination.write_text(jinja_env.get_template(f"{out_name}.j2").render(**common_vars))
|
|
print(f" Rendered {out_name}.j2 -> {destination}")
|
|
|
|
|
|
def git_auth_env(token):
|
|
"""Return credential-safe Git authentication shared by clone and push."""
|
|
auth_env = os.environ.copy()
|
|
auth_env["CI_BOT_TOKEN"] = token
|
|
auth_env["GIT_ASKPASS"] = str(Path(__file__).with_name("git-askpass.sh"))
|
|
auth_env["GIT_TERMINAL_PROMPT"] = "0"
|
|
return auth_env
|
|
|
|
|
|
def commit_and_push(apps_dir, message, token=None):
|
|
run(["git", "-C", str(apps_dir), "add", "-A"])
|
|
result = subprocess.run(["git", "-C", str(apps_dir), "diff", "--cached", "--quiet"], check=False)
|
|
if result.returncode == 0:
|
|
print("No manifest changes to commit")
|
|
return False
|
|
run(["git", "-C", str(apps_dir), "commit", "-m", message])
|
|
push_env = git_auth_env(token) if token else None
|
|
run(["git", "-C", str(apps_dir), "push"], env=push_env)
|
|
print("Manifests pushed — ArgoCD will sync")
|
|
return True
|