#!/bin/bash # detect-platform.sh - Detect git platform (Gitea or GitHub) for current repo # Usage: source detect-platform.sh && detect_platform # or: ./detect-platform.sh (prints platform name) detect_platform() { local remote_url remote_url=$(git remote get-url origin 2>/dev/null) if [[ -z "$remote_url" ]]; then echo "error: not a git repository or no origin remote" >&2 return 1 fi # Check for GitHub if [[ "$remote_url" == *"github.com"* ]]; then PLATFORM="github" export PLATFORM echo "github" return 0 fi # Check for common Gitea indicators # Gitea URLs typically don't contain github.com, gitlab.com, bitbucket.org if [[ "$remote_url" != *"gitlab.com"* ]] && \ [[ "$remote_url" != *"bitbucket.org"* ]]; then # Assume Gitea for self-hosted repos PLATFORM="gitea" export PLATFORM echo "gitea" return 0 fi PLATFORM="unknown" export PLATFORM echo "unknown" return 1 } get_repo_info() { local remote_url remote_url=$(git remote get-url origin 2>/dev/null) if [[ -z "$remote_url" ]]; then echo "error: not a git repository or no origin remote" >&2 return 1 fi # Extract owner/repo from URL # Handles: git@host:owner/repo.git, https://host/owner/repo.git, https://host/owner/repo local repo_path if [[ "$remote_url" == git@* ]]; then repo_path="${remote_url#*:}" else repo_path="${remote_url#*://}" repo_path="${repo_path#*/}" fi # Remove .git suffix if present repo_path="${repo_path%.git}" echo "$repo_path" } get_repo_owner() { local repo_info repo_info=$(get_repo_info) echo "${repo_info%%/*}" } get_repo_name() { local repo_info repo_info=$(get_repo_info) echo "${repo_info##*/}" } get_repo_slug() { get_repo_info } gitea_url_matches_host() { local url="${1:-}" host="${2:-}" [[ -n "$url" && -n "$host" ]] || return 1 python3 - "$url" "$host" <<'PY' import sys from urllib.parse import urlparse url, remote_host = sys.argv[1:] configured = urlparse(url) remote = urlparse(f"//{remote_host}") if configured.scheme not in {"http", "https"} or configured.hostname != remote.hostname: raise SystemExit(1) # Normalize by scheme: an implicit (portless) HTTP(S) URL and its explicit # default-port form (":80" for http, ":443" for https) name the same # provider endpoint. Apply that equivalence symmetrically -- whichever side # omits the port is treated as carrying the scheme's default port -- so # "configured implicit vs. remote explicit" and "configured explicit vs. # remote implicit" both match. (The remote side here is always an HTTP(S) # authority; an SSH remote's transport port is stripped by get_remote_host # before reaching this comparison, since it identifies an unrelated # service on the same host, not the HTTP(S) provider port.) default_port = 80 if configured.scheme == "http" else 443 normalized_configured = configured.port if configured.port is not None else default_port normalized_remote = remote.port if remote.port is not None else default_port if normalized_configured != normalized_remote: raise SystemExit(1) raise SystemExit(0) PY } get_gitea_service_for_host() { local host="$1" local cred_file="${MOSAIC_CREDENTIALS_FILE:-}" if [[ -z "$cred_file" ]]; then # Same resolution chain as _lib/credentials.sh: profile HOME, then # host-level /etc only if it exists; neither existing keeps the # $HOME default (matches the lib's final := fallback). cred_file="$HOME/.config/mosaic/credentials.json" if [[ ! -f "$cred_file" && -f /etc/mosaic/credentials.json ]]; then cred_file="/etc/mosaic/credentials.json" fi fi case "$host" in git.mosaicstack.dev) echo "mosaicstack" return 0 ;; git.uscllc.com) echo "usc" return 0 ;; esac [[ -f "$cred_file" ]] || return 1 command -v jq >/dev/null 2>&1 || return 1 jq -r --arg host "$host" ' .gitea // {} | to_entries[] | select((.value.url // "" | sub("/+$"; "")) | test("https?://" + $host + "$")) | .key ' "$cred_file" | head -n 1 } find_tea_login_for_host() { local host="$1" local logins_json command -v tea >/dev/null 2>&1 || return 1 logins_json=$(tea login list --output json 2>/dev/null) || return 1 TEA_LOGINS_JSON="$logins_json" python3 - "$host" <<'PY' import json import os import sys from urllib.parse import urlparse host = sys.argv[1] try: logins = json.loads(os.environ.get("TEA_LOGINS_JSON", "[]")) except Exception: raise SystemExit(1) for login in logins if isinstance(logins, list) else []: url = str(login.get("url") or login.get("URL") or "") name = str(login.get("name") or login.get("Name") or "") parsed = urlparse(url) if parsed.hostname == host and name: print(name) raise SystemExit(0) raise SystemExit(1) PY } tea_login_matches_host() { local login_name="$1" host="$2" local logins_json command -v tea >/dev/null 2>&1 || return 1 logins_json=$(tea login list --output json 2>/dev/null) || return 1 TEA_LOGINS_JSON="$logins_json" python3 - "$login_name" "$host" <<'PY' import json import os import sys from urllib.parse import urlparse login_name, host = sys.argv[1], sys.argv[2] try: logins = json.loads(os.environ.get("TEA_LOGINS_JSON", "[]")) except Exception: raise SystemExit(1) for login in logins if isinstance(logins, list) else []: url = str(login.get("url") or login.get("URL") or "") name = str(login.get("name") or login.get("Name") or "") parsed = urlparse(url) if name == login_name and parsed.hostname == host: raise SystemExit(0) raise SystemExit(1) PY } # Emit an actionable diagnostic to stderr when no tea login resolves for a host. # Callers that have a working API fallback may ignore the non-zero return of # get_gitea_login_for_host; this turns the previously SILENT failure into a loud, # greppable hint (available logins + override + add-login instructions). Printed to # stderr only, so it never contaminates stdout (the resolved login name) or log # assertions that capture tea/curl invocations. print_gitea_login_diagnostic() { local host="${1:-}" local available available=$( command -v tea >/dev/null 2>&1 || { echo "(tea CLI not installed)"; exit 0; } logins_json=$(tea login list --output json 2>/dev/null) || { echo "(could not query tea login list)"; exit 0; } TEA_LOGINS_JSON="$logins_json" python3 - <<'PY' import json, os from urllib.parse import urlparse try: logins = json.loads(os.environ.get("TEA_LOGINS_JSON", "[]")) except Exception: logins = [] rows = [] for login in logins if isinstance(logins, list) else []: name = str(login.get("name") or login.get("Name") or "") url = str(login.get("url") or login.get("URL") or "") host = urlparse(url).hostname or "?" if name: rows.append(f"{name} (host: {host})") print("; ".join(rows) if rows else "(none configured)") PY ) { echo "Error: no Gitea tea login matches host '$host'." echo " Available tea logins: ${available}" echo " Fix: set GITEA_LOGIN to a login whose URL host is '$host'," echo " or add one: tea login add --name --url https://$host --token " } >&2 } get_gitea_login_for_host() { local host="${1:-}" local login if [[ -z "$host" ]]; then host=$(get_remote_host) || return 1 fi if [[ -n "${GITEA_LOGIN:-}" ]]; then if tea_login_matches_host "$GITEA_LOGIN" "$host"; then echo "$GITEA_LOGIN" return 0 fi fi login=$(find_tea_login_for_host "$host" || true) if [[ -n "$login" ]]; then echo "$login" return 0 fi print_gitea_login_diagnostic "$host" return 1 } # Validate the current authenticated Gitea user for a resolved Tea login. # Tea stores a user name with each login which can become stale after user rename, # token rotation, or server migration. Querying /user derives the identity from the # active credential instead of trusting that saved name. Callers fall back to the # host-scoped API path when this validation fails. get_gitea_authenticated_user() { local login_name="$1" response command -v tea >/dev/null 2>&1 || return 1 response=$(tea api --login "$login_name" /user 2>/dev/null) || return 1 TEA_AUTHENTICATED_USER_JSON="$response" python3 - <<'PY' import json import os try: user = json.loads(os.environ["TEA_AUTHENTICATED_USER_JSON"]) except (KeyError, json.JSONDecodeError): raise SystemExit(1) login = user.get("login") if isinstance(user, dict) else None if isinstance(login, str) and login: print(login) raise SystemExit(0) raise SystemExit(1) PY } get_default_tea_login() { local logins_json command -v tea >/dev/null 2>&1 || return 1 logins_json=$(tea login list --output json 2>/dev/null) || return 1 TEA_LOGINS_JSON="$logins_json" python3 - <<'PY' import json import os try: logins = json.loads(os.environ.get("TEA_LOGINS_JSON", "[]")) except Exception: raise SystemExit(1) if not isinstance(logins, list) or not logins: raise SystemExit(1) for login in logins: if not isinstance(login, dict): continue is_default = str(login.get("default") or login.get("Default") or "").lower() name = str(login.get("name") or login.get("Name") or "") if name and is_default == "true": print(name) raise SystemExit(0) for login in logins: if not isinstance(login, dict): continue name = str(login.get("name") or login.get("Name") or "") if name: print(name) raise SystemExit(0) raise SystemExit(1) PY } get_gitea_login_for_repo_override() { local login if [[ -n "${GITEA_LOGIN:-}" ]]; then echo "$GITEA_LOGIN" return 0 fi login=$(get_default_tea_login || true) if [[ -n "$login" ]]; then echo "$login" return 0 fi return 1 } get_host_from_url() { local url="${1:-}" [[ -n "$url" ]] || return 1 python3 - "$url" <<'PY' import sys from urllib.parse import urlparse parsed = urlparse(sys.argv[1]) if parsed.hostname: print(parsed.hostname) raise SystemExit(0) raise SystemExit(1) PY } get_gitea_api_host_for_repo_override() { if [[ -n "${GITEA_HOST:-}" ]]; then echo "$GITEA_HOST" return 0 fi get_host_from_url "${GITEA_URL:-}" } # Resolve owner/repo relative to a configured Gitea base URL. HTTP(S) clone # URLs can include the deployment prefix (for example /gitea/owner/repo.git), # but Gitea's /repos API expects only owner/repo. Root-mounted and SSH clone # forms retain their existing owner/repo behavior. get_gitea_repo_slug_for_url() { local configured_url="$1" remote_url remote_url=$(git remote get-url origin 2>/dev/null) || return 1 if [[ "$remote_url" =~ ^https?:// ]]; then python3 - "$remote_url" "$configured_url" <<'PY' import sys from urllib.parse import urlparse remote = urlparse(sys.argv[1]) base = urlparse(sys.argv[2]) remote_path = remote.path.strip("/") if remote_path.endswith(".git"): remote_path = remote_path[:-4] base_path = base.path.strip("/") remote_parts = [part for part in remote_path.split("/") if part] base_parts = [part for part in base_path.split("/") if part] if base_parts and remote_parts[:len(base_parts)] == base_parts: repo_parts = remote_parts[len(base_parts):] elif len(remote_parts) == 2: # Preserve a root-shaped clone URL when provider API configuration carries # a reverse-proxy prefix separately. repo_parts = remote_parts else: raise SystemExit(1) if len(repo_parts) != 2: raise SystemExit(1) print("/".join(repo_parts)) PY return fi get_repo_slug } get_gitea_repo_args() { local repo host login repo=$(get_repo_slug) || return 1 host=$(get_remote_host) || return 1 login=$(get_gitea_login_for_host "$host") || return 1 printf -- '--repo %q --login %q' "$repo" "$login" } get_gitea_login() { get_gitea_login_for_host "$(get_remote_host)" } get_remote_host() { local remote_url remote_url=$(git remote get-url origin 2>/dev/null || true) if [[ -z "$remote_url" ]]; then return 1 fi if [[ "$remote_url" =~ ^https?://([^/]+)/ ]]; then local host="${BASH_REMATCH[1]}" echo "${host##*@}" return 0 fi if [[ "$remote_url" =~ ^ssh://([^/]+)/ ]]; then local host="${BASH_REMATCH[1]}" host="${host##*@}" # Strip an SSH transport port (e.g. "git.example:2222"): it names the # SSH daemon port, not the HTTP(S) provider API port, and must not # feed gitea_url_matches_host's port comparison (#850). echo "${host%%:*}" return 0 fi if [[ "$remote_url" =~ ^git@([^:]+): ]]; then echo "${BASH_REMATCH[1]}" return 0 fi return 1 } # Resolve the configured Gitea base URL for a host from the same credential # source used by get_gitea_token. The scheme and any deployment path prefix are # provider configuration and must not be reconstructed from the git remote. get_gitea_url_for_host() { local host="$1" script_dir cred_loader url script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" cred_loader="$script_dir/../_lib/credentials.sh" if [[ -f "$cred_loader" ]]; then url=$( # shellcheck source=/dev/null source "$cred_loader" unset GITEA_TOKEN GITEA_URL case "$host" in git.mosaicstack.dev) load_credentials gitea-mosaicstack 2>/dev/null ;; git.uscllc.com) load_credentials gitea-usc 2>/dev/null ;; *) for svc in gitea-mosaicstack gitea-usc; do unset GITEA_TOKEN GITEA_URL load_credentials "$svc" 2>/dev/null || continue if gitea_url_matches_host "${GITEA_URL:-}" "$host"; then break fi unset GITEA_TOKEN GITEA_URL done ;; esac if gitea_url_matches_host "${GITEA_URL:-}" "$host"; then printf '%s' "${GITEA_URL%/}" fi ) if [[ -n "$url" ]]; then printf '%s\n' "$url" return 0 fi fi if gitea_url_matches_host "${GITEA_URL:-}" "$host"; then printf '%s\n' "${GITEA_URL%/}" return 0 fi return 1 } # Resolve a Gitea API token for the given host. # Priority: Mosaic credential loader → GITEA_TOKEN env → ~/.git-credentials get_gitea_token() { local host="$1" local script_dir script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" local cred_loader="$script_dir/../_lib/credentials.sh" # 1. Mosaic credential loader (host → service mapping, run in subshell to avoid polluting env) if [[ -f "$cred_loader" ]]; then local token token=$( # shellcheck source=/dev/null source "$cred_loader" # Host-specific wrapper resolution must not inherit caller/global GITEA_*. # load_credentials intentionally preserves existing env vars for interactive use, # but metadata/merge wrappers need credentials matching the remote host. unset GITEA_TOKEN GITEA_URL case "$host" in git.mosaicstack.dev) load_credentials gitea-mosaicstack 2>/dev/null ;; git.uscllc.com) load_credentials gitea-usc 2>/dev/null ;; *) local matched=false for svc in gitea-mosaicstack gitea-usc; do unset GITEA_TOKEN GITEA_URL load_credentials "$svc" 2>/dev/null || continue if gitea_url_matches_host "${GITEA_URL:-}" "$host"; then matched=true break fi done if [[ "$matched" != true ]]; then unset GITEA_TOKEN GITEA_URL fi ;; esac echo "${GITEA_TOKEN:-}" ) if [[ -n "$token" ]]; then echo "$token" return 0 fi fi # 2. GITEA_TOKEN env var (only when GITEA_URL, if present, matches the remote host) if [[ -n "${GITEA_TOKEN:-}" ]]; then if [[ -z "${GITEA_URL:-}" ]] || gitea_url_matches_host "$GITEA_URL" "$host"; then echo "$GITEA_TOKEN" return 0 fi fi # 3. ~/.git-credentials file local creds="$HOME/.git-credentials" if [[ -f "$creds" ]]; then local token token=$(grep -F "$host" "$creds" 2>/dev/null | sed -n 's#https\?://[^@]*:\([^@/]*\)@.*#\1#p' | head -n 1) if [[ -n "$token" ]]; then echo "$token" return 0 fi fi return 1 } # Stage the Gitea bearer credential for curl OUTSIDE the process argument vector. # Passing "-H 'Authorization: token '" on the curl command line exposes # the token to anyone who can read the process table (ps / /proc//cmdline) # for the lifetime of the request. Instead, write the header into a private # (mode 0600) curl config file and have callers pass it with `curl --config`, so # only the FILE PATH — never the token — appears in argv. Prints the temp file # path on success; the caller OWNS the file and MUST remove it on every exit # path (success and failure). $1 = bearer token. Callers must not log the token. gitea_write_auth_config() { local token="$1" auth_file auth_file=$(mktemp "${TMPDIR:-/tmp}/mosaic-gitea-auth.XXXXXX") || return 1 # mktemp already creates the file with 0600; be explicit in case of an # unusual umask so the credential is never briefly group/other readable. chmod 600 "$auth_file" 2>/dev/null || true # `header = "..."` is curl's config syntax for an extra request header. Only # this filename reaches curl's argv; the token stays on disk, readable solely # by this user, and is unlinked by the caller's trap after the request. if ! printf 'header = "Authorization: token %s"\n' "$token" > "$auth_file"; then rm -f "$auth_file" return 1 fi printf '%s' "$auth_file" } # Resolve the API token for a SPECIFIC tea login name from tea's own config # (the same store tea itself writes/reads for `--login `). This is what # lets a REST write be performed AS the selected --login identity: tea keys its # per-login tokens by `name` in $XDG_CONFIG_HOME/tea/config.yml (default # ~/.config/tea/config.yml), exactly as the `tea` CLI resolves them, so a # --login override and its REST read-back bind to the SAME credential/identity. # # $2 (repo host) binds the selected credential to the TARGET host: a tea login # also records the `url` it authenticates against, and the matched login's URL # host MUST equal the repo host. This fails closed when an override login is # configured for a DIFFERENT host than the repo remote, so a login name shared # across hosts (or a mistargeted override) can never send one host's credential # to another host (cross-host credential leak). When $2 is empty the host bind # is skipped (host-agnostic lookup) — callers that write should always pass it. # # Prints the token on success; returns non-zero (no output) if the config, a # matching login token, or the host bind cannot be satisfied. Callers must not # log the result. get_gitea_token_for_login() { local login_name="$1" repo_host="${2:-}" config_file [[ -n "$login_name" ]] || return 1 config_file="${XDG_CONFIG_HOME:-$HOME/.config}/tea/config.yml" [[ -f "$config_file" ]] || return 1 LOGIN_NAME="$login_name" REPO_HOST="$repo_host" python3 - "$config_file" <<'PY' import datetime import os import re import sys from urllib.parse import urlparse wanted = os.environ["LOGIN_NAME"] repo_host = os.environ.get("REPO_HOST", "").strip().lower() config_path = sys.argv[1] # PyYAML 6.0.3 SafeLoader implicit resolver patterns (YAML 1.1). An UNQUOTED # plain scalar matching any of these is resolved by PyYAML to a NON-string type # (null->None, bool, int, float, timestamp->date/datetime); a quoted scalar is # ALWAYS a string. These mirror yaml/resolver.py so the PyYAML-absent fallback # types unquoted scalars exactly as PyYAML would (see _implicit_nonstring). _IMPLICIT_NULL = re.compile(r"^(?:~|null|Null|NULL|)$") _IMPLICIT_BOOL = re.compile( r"^(?:yes|Yes|YES|no|No|NO|true|True|TRUE|false|False|FALSE" r"|on|On|ON|off|Off|OFF)$" ) _IMPLICIT_INT = re.compile( r"^(?:[-+]?0b[0-1_]+" r"|[-+]?0[0-7_]+" r"|[-+]?(?:0|[1-9][0-9_]*)" r"|[-+]?0x[0-9a-fA-F_]+" r"|[-+]?[1-9][0-9_]*(?::[0-5]?[0-9])+)$" ) # NOTE: the exponent group requires an EXPLICIT sign ([eE][-+][0-9]+), matching # PyYAML 6.0.3's resolver.py float regex exactly. An unsigned exponent (e.g. # "1.0e10", ".5e10", "4.e8") is NOT matched by PyYAML's implicit float resolver # -- PyYAML resolves those as plain strings -- so this pattern must not match # them either, or the fallback over-rejects a token PyYAML would emit verbatim. _IMPLICIT_FLOAT = re.compile( r"^(?:[-+]?(?:[0-9][0-9_]*)\.[0-9_]*(?:[eE][-+][0-9]+)?" r"|\.[0-9][0-9_]*(?:[eE][-+][0-9]+)?" r"|[-+]?[0-9][0-9_]*(?::[0-5]?[0-9])+\.[0-9_]*" r"|[-+]?\.(?:inf|Inf|INF)" r"|\.(?:nan|NaN|NAN))$" ) _IMPLICIT_TIMESTAMP = re.compile( r"^(?:[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]" r"|[0-9][0-9][0-9][0-9]-[0-9][0-9]?-[0-9][0-9]?" r"(?:[Tt]|[ \t]+)[0-9][0-9]?" r":[0-9][0-9]:[0-9][0-9](?:\.[0-9]*)?" r"(?:[ \t]*(?:Z|[-+][0-9][0-9]?(?::[0-9][0-9])?))?)$" ) def _implicit_nonstring(text): # True when an UNQUOTED plain scalar would be resolved by PyYAML's SafeLoader # to a non-string type (null/bool/int/float/timestamp). Fuzzed against real # PyYAML 6.0.3: it never returns False where PyYAML types the scalar as a # non-string (i.e. never fail-open), and (post exponent-sign fix) no longer # over-rejects unsigned-exponent spellings like "4.e8" or "1.0e10" -- those # match PyYAML's own implicit float resolver exactly (explicit sign only), # so PyYAML keeps them as strings and this function now agrees. return bool( _IMPLICIT_NULL.match(text) or _IMPLICIT_BOOL.match(text) or _IMPLICIT_INT.match(text) or _IMPLICIT_FLOAT.match(text) or _IMPLICIT_TIMESTAMP.match(text) ) # PyYAML 6.0.3 SafeConstructor timestamp regexp (yaml/constructor.py). The # constructor RE-parses a timestamp-tagged scalar with THIS pattern and then # builds a datetime.date/datetime, which raises ValueError for an out-of-range # calendar field (e.g. month 99, hour 25). _IMPLICIT_TIMESTAMP (the RESOLVER # pattern) is byte-identical to PyYAML's resolver, so anything it tags is also # tagged by PyYAML and re-matched here. _TIMESTAMP_CONSTRUCT = re.compile( r"""^(?P[0-9][0-9][0-9][0-9]) -(?P[0-9][0-9]?) -(?P[0-9][0-9]?) (?:(?:[Tt]|[ \t]+) (?P[0-9][0-9]?) :(?P[0-9][0-9]) :(?P[0-9][0-9]) (?:\.(?P[0-9]*))? (?:[ \t]*(?PZ|(?P[-+])(?P[0-9][0-9]?) (?::(?P[0-9][0-9]))?))?)?$""", re.X, ) def _int_constructible(text): # Replicate PyYAML SafeConstructor.construct_yaml_int and report whether it # would succeed. A resolver-tagged int whose radix body is empty after # underscore removal (e.g. "0b_", "0x_", "0x__") makes int(base) raise, so # PyYAML fails the WHOLE document -> the fallback must fail closed too. value = text.replace("_", "") if value[:1] in ("+", "-"): value = value[1:] if value == "0": return True try: if value.startswith("0b"): int(value[2:], 2) elif value.startswith("0x"): int(value[2:], 16) elif value[:1] == "0": int(value, 8) elif ":" in value: [int(part) for part in value.split(":")] else: int(value) except ValueError: return False return True def _float_constructible(text): # Replicate PyYAML SafeConstructor.construct_yaml_float. Retained for the # WHOLE-document invariant: _IMPLICIT_FLOAT now matches PyYAML's resolver # pattern exactly (explicit-sign exponent only), and every scalar it tags # is float()-constructible, so this never fails closed where PyYAML would # emit a token. value = text.replace("_", "").lower() if value[:1] in ("+", "-"): value = value[1:] if value in (".inf", ".nan"): return True try: if ":" in value: [float(part) for part in value.split(":")] else: float(value) except ValueError: return False return True def _timestamp_constructible(text): # Replicate PyYAML SafeConstructor.construct_yaml_timestamp: build the same # datetime.date/datetime and report whether it raises. Returns False for an # out-of-range calendar field (month/day/hour/...), matching PyYAML's # whole-document ValueError. match = _TIMESTAMP_CONSTRUCT.match(text) if not match: return False values = match.groupdict() try: year = int(values["year"]) month = int(values["month"]) day = int(values["day"]) if not values["hour"]: datetime.date(year, month, day) return True hour = int(values["hour"]) minute = int(values["minute"]) second = int(values["second"]) fraction = 0 if values["fraction"]: frac = values["fraction"][:6] frac += "0" * (6 - len(frac)) fraction = int(frac) tzinfo = None if values["tz_sign"]: tz_hour = int(values["tz_hour"]) tz_minute = int(values["tz_minute"] or 0) delta = datetime.timedelta(hours=tz_hour, minutes=tz_minute) if values["tz_sign"] == "-": delta = -delta tzinfo = datetime.timezone(delta) elif values["tz"]: tzinfo = datetime.timezone.utc datetime.datetime( year, month, day, hour, minute, second, fraction, tzinfo=tzinfo ) except (ValueError, OverflowError): return False return True def _constructible(text): # WHOLE-DOCUMENT INVARIANT: the fallback resolves the SAME login token as # PyYAML safe_load or fails closed -- never less conservative -- INCLUDING # when PyYAML raises a CONSTRUCTOR error anywhere in the document. A plain # scalar can match a typed implicit resolver (int/float/timestamp) yet NOT be # constructible (e.g. 2023-99-99, 0b_, 0x_); PyYAML then raises on the whole # load and yields no token, so the fallback MUST fail closed for the whole # document too. This returns True only when PyYAML's constructor would build # the scalar (null/bool token sets are always constructible), else False so # the caller fails closed. `text` is assumed to satisfy _implicit_nonstring. if _IMPLICIT_NULL.match(text) or _IMPLICIT_BOOL.match(text): return True if _IMPLICIT_TIMESTAMP.match(text): return _timestamp_constructible(text) if _IMPLICIT_INT.match(text): return _int_constructible(text) if _IMPLICIT_FLOAT.match(text): return _float_constructible(text) return True # PyYAML's Reader.check_printable scans the ENTIRE raw input stream (not just # scalar contents, and NOT scoped by quoting) for code points outside its # printable set and raises ReaderError -- a whole-document reject -- the instant # one is found, no matter where it sits (an unrelated field, a comment, inside or # outside quotes). To guarantee parity WITHOUT a hand-rolled subset (which missed # the C1 block and BMP noncharacters), this is PyYAML 6.0.3's EXACT Reader # printable definition, negated verbatim: # Reader.NON_PRINTABLE = # re.compile('[^\x09\x0A\x0D\x20-\x7E\x85\xA0-퟿-�' # '\U00010000-\U0010FFFF]') # i.e. PRINTABLE = {TAB(0x09), LF(0x0A), CR(0x0D), 0x20-0x7E, NEL(0x85), # 0xA0-0xD7FF, 0xE000-0xFFFD, 0x10000-0x10FFFF}; EVERYTHING else (all other C0 # controls, DEL 0x7F, the entire C1 block 0x80-0x84/0x86-0x9F, the surrogate # range 0xD800-0xDFFF, and the BMP noncharacters 0xFFFE/0xFFFF) is non-printable # -> whole-document fail closed. Verified empirically against real PyYAML 6.0.3 # in both directions: the C1 bytes and 0xFFFE/0xFFFF reject; NEL(0x85), 0xA0, and # astral code points (e.g. U+1F600) are accepted and resolve the token. The prior # C0/DEL/tab behavior is a strict subset of this. (A surrogate code point cannot # occur in valid UTF-8, so it also trips the UTF-8 decode on read and fails # closed; it is included here for exact definitional parity.) _FORBIDDEN_CONTROL = re.compile( "[^\x09\x0a\x0d\x20-\x7e\x85\xa0-퟿-�\U00010000-\U0010ffff]" ) # Sentinel: "outside the supported subset -> fail closed". Distinct from a # genuine null (None), which is a valid resolved value. _FAIL = object() # Separators are SPACE-only: PyYAML rejects a tab used as key/value whitespace # (before or after the ':') with a scanner error, so a tab there must NOT be # treated as a benign separator. Space before the colon and one space after it # stay valid (PyYAML strips a plain key's trailing spaces); a tab in either # position makes the whole line fail to match -> the caller fails closed. _KEY_RE = re.compile(r"^([A-Za-z0-9_][A-Za-z0-9_.\-]*) *:(?:[ ](.*)|)$") _FLOW = set("[]{}*&!") class _Bail(Exception): # Raised the instant the document leaves the narrow tea-config subset this # recognizer can prove it resolves IDENTICALLY to PyYAML. Caught by # _safe_parse, which then fails closed (returns _FAIL) rather than guess. pass def _scalar(raw): # Resolve a single flow scalar (quoted or plain) the way PyYAML would for # tea's simple values, or _FAIL when it is outside the supported subset so # the caller fails closed instead of guessing. # # A quoted scalar is ALWAYS a string (its contents returned verbatim); a '#' # inside quotes is data. An UNQUOTED scalar ends at the first whitespace # -preceded '#' (a YAML comment must be preceded by whitespace or line start, # so "abc#def" stays literal while "abc # note" becomes "abc"), and is then # subject to PyYAML's implicit typing: forms like 12345 / null / ~ / yes / # 3.14 / a timestamp resolve to a NON-string (int/None/bool/float/date), so # they return None here (PyYAML's path rejects a non-str token via _accept). # Strip SPACES only, never tabs: PyYAML raises a scanner error on a tab in a # plain/leading/trailing scalar position, so a tab must be PRESERVED here to # trip the fail-closed guard below rather than be silently normalized away. value = raw.strip(" ") if not value: return None # empty plain scalar -> null if value[0] in ("'", '"'): quote = value[0] end = value.find(quote, 1) if end == -1: return _FAIL # unterminated quote: PyYAML would error / continue rest = value[end + 1:].strip(" ") if rest and not rest.startswith("#"): return _FAIL # trailing junk after a quoted scalar inner = value[1:end] # Single-quote '' escaping and double-quote backslash escapes are NOT # interpreted here; reject any scalar that uses them so we never diverge # from PyYAML on escape handling. if quote == "'" and "'" in inner: return _FAIL if quote == '"' and "\\" in inner: return _FAIL return inner for i, ch in enumerate(value): if ch == "#" and (i == 0 or value[i - 1] in (" ", "\t")): value = value[:i] break value = value.strip(" ") # spaces only; a tab must survive to fail closed if not value: return None if value[0] in _FLOW or value[0] in ("|", ">", "@", "`", '"', "'", "%", ","): return _FAIL # flow / block-scalar / reserved / directive / anchor / quote # NOTE: "?" is deliberately NOT in the blanket-reject set above. Empirically # verified against real PyYAML 6.0.3: "?x" (indicator immediately followed by # a non-space) is a plain scalar string "?x" like "-x" and ":x" are -- only # "?" alone or "? " (followed by space/EOL) opens a complex mapping key and # is illegal in a value position. Blanket-rejecting every leading "?" would # over-reject a token PyYAML accepts verbatim; the narrower check below # (mirroring "-" and ":") handles exactly the illegal bare-indicator forms. if value[0] in ("-", "?", ":") and (len(value) == 1 or value[1] in (" ", "\t")): # A bare block indicator, not a plain scalar: '-'/'- ' opens a sequence # entry, '?'/'? ' a complex mapping key, ':'/': ' a mapping value -- all # illegal in a value position, where PyYAML raises a scanner error on the # whole document. "-x"/"-1"/"?x"/":x" (indicator NOT followed by space) # remain valid plain scalars and fall through. Fail closed on the bare # indicator so the fallback never emits a token PyYAML would refuse. return _FAIL # NO blanket internal-indicator reject here. In BLOCK context (where a tea # config value always sits) PyYAML treats ',[]{}' as ordinary plain-scalar # content -- they are flow indicators ONLY inside a flow collection -- and # treats '!&*|>#%@`' plus quotes as significant ONLY at the FIRST non-space # character of a node (a leading one starts a tag/anchor/alias/block-scalar/ # comment/directive/reserved/quote), which the leading-char guard above # (value[0] ...) already rejects. Verified empirically against real PyYAML # 6.0.3: an INTERNAL '!' ',' '[' ']' '{' '}' '&' '*' '#'(not space-preceded) # ':'(not space-followed) all resolve as a plain-string token. The only # internal positions PyYAML treats as significant in block context are ' #' # (whitespace-preceded '#' -> comment; stripped by the loop above) and ': ' # or a trailing ':' (-> mapping; rejected by the ": "/endswith(":") guard # below). A blanket any(ch in _FLOW ...) scan over-rejected those internal # indicators, dropping a token PyYAML emits verbatim; removing it restores # parity while the position-specific guards keep the fail-open direction shut. if "\t" in value: # A tab anywhere in a plain scalar (leading, trailing, or embedded) is a # PyYAML scanner error on the whole document -- it accepts tabs ONLY # inside quoted scalars (handled above, returned verbatim). Fail closed # so the fallback never emits a token PyYAML would refuse over a tab. return _FAIL if ": " in value or value.endswith(":"): return _FAIL # nested-mapping-in-scalar / ambiguous if _implicit_nonstring(value): # A plain scalar PyYAML would tag as a non-string (null/bool/int/float/ # timestamp). If PyYAML's CONSTRUCTOR would build it, the value is a # non-string -> null-equivalent for a token field: return None and keep # parsing (as before). But if it matches a typed implicit resolver yet is # NOT constructible (e.g. 2023-99-99, 0b_, 0x_), PyYAML raises on the # WHOLE document and yields no token, so the fallback MUST fail closed # for the whole document too -> _FAIL (which the caller turns into _Bail). if not _constructible(value): return _FAIL return None return value class _Parser: # A deliberately NARROW, conservative recognizer for the block-style YAML # subset tea writes (mappings of scalar fields; a `logins:` block SEQUENCE of # such mappings; optional shallow nested mappings for e.g. preferences). It # reconstructs the SAME Python object PyYAML's SafeLoader would, but the # instant it meets anything it cannot prove it handles identically -- a # document marker (--- / ...), a block scalar (| / >), a flow collection, a # duplicate mapping key, inconsistent indentation, an escape, or any line # outside the grammar -- it raises _Bail so the whole resolution fails # closed. This guarantees the module invariant (only ever MORE conservative # than PyYAML, never less) at the DOCUMENT level, closing the structural # fail-open classes (nested-logins shadow, block-scalar shadow, duplicate # root key, malformed-after-valid, and extra-document) that a line scan that # does not validate whole-document structure would miss. def __init__(self, lines): self.toks = [] for raw in lines: if not raw.strip(): continue lead = raw[: len(raw) - len(raw.lstrip(" \t"))] if "\t" in lead: raise _Bail() # tab in indentation: PyYAML scanner error indent = len(lead) body = raw[indent:] if body.lstrip().startswith("#"): continue if re.match(r"^(---|\.\.\.)(\s|$)", body) or body in ("---", "..."): raise _Bail() # document / end marker -> multi-doc -> fail closed # rstrip SPACES only: a trailing tab is a PyYAML scanner error, so it # must be kept on the token body to reach the fail-closed guards # (rstrip() would swallow it and let a bad line resolve a token). self.toks.append((indent, body.rstrip(" "))) self.i = 0 def peek(self): return self.toks[self.i] if self.i < len(self.toks) else None def parse_document(self): if not self.toks: return None node = self.parse_node(0) if self.i != len(self.toks): raise _Bail() # trailing unconsumed content -> malformed return node def parse_node(self, min_indent): tok = self.peek() if tok is None: return None indent, body = tok if indent < min_indent: return None if body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t")): return self.parse_seq(indent) return self.parse_map(indent) def parse_map(self, indent): result = {} while True: tok = self.peek() if tok is None: break cur_indent, body = tok if cur_indent < indent: break if cur_indent > indent: raise _Bail() # unexpected deeper line (bad indentation) if body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t")): raise _Bail() # sequence item where a mapping entry was expected m = _KEY_RE.match(body) if not m: raise _Bail() key = m.group(1) inline = m.group(2) self.i += 1 if key in result: raise _Bail() # duplicate mapping key (PyYAML last-wins; we bail) if inline is not None and inline.strip(" ") != "": val = _scalar(inline) if val is _FAIL: raise _Bail() result[key] = val else: result[key] = self.parse_block_value(indent) return result def parse_block_value(self, key_indent): # The value after a "key:" with no inline scalar. A block SEQUENCE may sit # at the same indent as the key (YAML permits `- ` aligned with the key -- # tea's own on-disk shape) or deeper; a block MAPPING must be strictly # deeper; otherwise the value is null. nxt = self.peek() if nxt is None: return None ni, nb = nxt is_item = nb.startswith("-") and (len(nb) == 1 or nb[1] in (" ", "\t")) if is_item and ni >= key_indent: return self.parse_seq(ni) if ni > key_indent: return self.parse_node(ni) return None def parse_seq(self, indent): result = [] while True: tok = self.peek() if tok is None: break cur_indent, body = tok if cur_indent < indent: break if cur_indent > indent: raise _Bail() if not (body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t"))): break # a mapping entry at this indent ends the sequence rest = body[1:].strip(" ") # spaces only; a tab must survive to bail self.i += 1 if rest == "": nxt = self.peek() if nxt is not None and nxt[0] > indent: result.append(self.parse_node(indent + 1)) else: result.append(None) continue km = _KEY_RE.match(rest) if km: # "- key: value" opens a mapping whose fields continue at the # column where the content after the dash began. field_indent = indent + (len(body) - len(body[1:].lstrip())) result.append(self.parse_inline_map(field_indent, km)) else: val = _scalar(rest) if val is _FAIL: raise _Bail() result.append(val) return result def parse_inline_map(self, field_indent, first_match): result = {} key = first_match.group(1) inline = first_match.group(2) if inline is not None and inline.strip(" ") != "": val = _scalar(inline) if val is _FAIL: raise _Bail() result[key] = val else: result[key] = self.parse_block_value(field_indent) while True: tok = self.peek() if tok is None: break cur_indent, body = tok if cur_indent != field_indent: if cur_indent > field_indent: raise _Bail() break if body.startswith("-") and (len(body) == 1 or body[1] in (" ", "\t")): raise _Bail() m = _KEY_RE.match(body) if not m: raise _Bail() k = m.group(1) iv = m.group(2) self.i += 1 if k in result: raise _Bail() if iv is not None and iv.strip(" ") != "": v = _scalar(iv) if v is _FAIL: raise _Bail() result[k] = v else: result[k] = self.parse_block_value(field_indent) return result def _safe_parse(lines): # Return the parsed root object (dict/list/scalar/None) when the WHOLE # document is inside the supported subset, else _FAIL (fail closed). try: return _Parser(lines).parse_document() except _Bail: return _FAIL except Exception: return _FAIL def _url_matches_repo_host(url): # Mirror gitea_url_matches_host (detect-platform.sh): the login's recorded # URL must name the SAME host AND the SAME effective port as the repo remote # host, not merely the same hostname. A login configured for an explicit, # non-default provider port (e.g. :9443) must NOT satisfy a portless (default # -port) repo host, and a login on the matching port (e.g. :8443) must NOT be # rejected. Ports are normalized by applying the login URL's scheme default # (80 for http, else 443) to whichever side omits the port, symmetrically, so # an implicit port and its explicit default-port form compare equal. if not isinstance(url, str) or not url: return False configured = urlparse(url if "//" in url else f"//{url}") remote = urlparse(f"//{repo_host}") configured_host = configured.hostname if not configured_host or configured_host.lower() != (remote.hostname or "").lower(): return False default_port = 80 if configured.scheme == "http" else 443 configured_port = configured.port if configured.port is not None else default_port remote_port = remote.port if remote.port is not None else default_port return configured_port == remote_port def _accept(token, url): # Enforce the host bind before surfacing a token. When a repo host is given, # the login's recorded URL host AND port must match it; a login with no # usable URL (or a mismatched host/port) is rejected (fail closed) so a # cross-host (or cross-port) credential is never emitted. if not isinstance(token, str) or not token: return None if repo_host: if not _url_matches_repo_host(url): return None return token def _token_via_pyyaml(): # Preferred, fully general path when PyYAML is installed. Raises ImportError # (caught by the caller) when the module is unavailable so the environment # -robust fallback can take over instead of failing closed on every host # that lacks PyYAML. import yaml with open(config_path, encoding="utf-8") as handle: config = yaml.safe_load(handle) logins = config.get("logins") if isinstance(config, dict) else None if not isinstance(logins, list): return None for login in logins: if isinstance(login, dict) and str(login.get("name") or "") == wanted: return _accept(login.get("token"), login.get("url")) return None def _token_via_lines(): # Conservative fallback for hosts without PyYAML. It parses config.yml with a # strict recognizer (_safe_parse) of the narrow block-style subset tea writes, # which reconstructs the SAME object PyYAML would OR fails closed (_FAIL) on # ANYTHING it cannot prove it resolves identically -- document markers, block # scalars, flow collections, duplicate keys, inconsistent indentation, or any # line outside the grammar. On a recognized document it then resolves the # login EXACTLY as the PyYAML path does (root `logins` list -> first entry # whose `name` equals the request -> host/port-bound token), so the fallback # can only ever be MORE conservative than PyYAML, never less. with open(config_path, encoding="utf-8") as handle: raw_text = handle.read() # Whole-document, position-independent reject: PyYAML's Reader rejects the # ENTIRE document (ReaderError) if a NON-printable code point (per its exact # printable definition -- see _FORBIDDEN_CONTROL: C0 controls, DEL, the C1 # block, surrogates, and 0xFFFE/0xFFFF) appears ANYWHERE in the raw stream -- # in a plain scalar, inside single- or double-quoted scalars, in a comment, # or in a field this recognizer never even looks at. Checking the raw text # (before splitlines(), which itself splits on some of these same code points # -- e.g. \x0b, \x0c, \x1c-\x1e, and NEL \x85 -- and would otherwise obscure # them) mirrors that exactly: fail closed for the whole document rather than # risk emitting a token PyYAML would have refused. (NEL 0x85 is PRINTABLE to # PyYAML's Reader, so it is NOT in _FORBIDDEN_CONTROL; where PyYAML treats it # as a line break the fallback's splitlines() agrees, and where PyYAML then # errors the recognizer's structural guards fail closed identically.) if _FORBIDDEN_CONTROL.search(raw_text): return None lines = raw_text.splitlines() config = _safe_parse(lines) if config is _FAIL: return None logins = config.get("logins") if isinstance(config, dict) else None if not isinstance(logins, list): return None for login in logins: if isinstance(login, dict) and str(login.get("name") or "") == wanted: return _accept(login.get("token"), login.get("url")) return None try: try: token = _token_via_pyyaml() except ImportError: token = _token_via_lines() except Exception: raise SystemExit(1) if isinstance(token, str) and token: print(token) raise SystemExit(0) raise SystemExit(1) PY } # Resolve HTTPS basic auth credentials for a Gitea host from ~/.git-credentials. # Prints "username:password" for direct curl -u consumption. Callers must not log it. get_gitea_basic_auth() { local host="$1" local creds="$HOME/.git-credentials" if [[ ! -f "$creds" ]]; then return 1 fi python3 - "$host" "$creds" <<'PY' import sys from pathlib import Path from urllib.parse import unquote, urlparse host = sys.argv[1] creds = Path(sys.argv[2]) for line in creds.read_text(encoding="utf-8").splitlines(): parsed = urlparse(line.strip()) if parsed.hostname != host: continue username = unquote(parsed.username or "") password = unquote(parsed.password or "") if username and password: print(f"{username}:{password}") raise SystemExit(0) raise SystemExit(1) PY } # If script is run directly (not sourced), output the platform if [[ "${BASH_SOURCE[0]}" == "${0}" ]]; then detect_platform fi