8ac7e70f0d
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
The PyYAML-absent fallback's _strip_properties consumed a plain anchor
('&name') per-scalar, with no document-scoped registry of declared anchor
names. Real PyYAML's Composer rejects a duplicate anchor NAME declared
anywhere in the document (ComposerError, whole-document fail), but the
fallback would still emit the later token: fail-open (HIGH). Fix: reject
every '&'-anchor property unconditionally instead of building a hand-rolled
duplicate registry, which guarantees zero fail-open by construction at the
cost of a deliberate, endorsed over-reject on a single non-duplicated
anchor. The '!' non-specific-tag branch is unchanged ('! x' -> 'x' still
resolves).
1468 lines
57 KiB
Bash
Executable File
1468 lines
57 KiB
Bash
Executable File
#!/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:-<unknown>}"
|
||
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 <name> --url https://$host --token <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 <value>'" on the curl command line exposes
|
||
# the token to anyone who can read the process table (ps / /proc/<pid>/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 <name>`). 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<year>[0-9][0-9][0-9][0-9])
|
||
-(?P<month>[0-9][0-9]?)
|
||
-(?P<day>[0-9][0-9]?)
|
||
(?:(?:[Tt]|[ \t]+)
|
||
(?P<hour>[0-9][0-9]?)
|
||
:(?P<minute>[0-9][0-9])
|
||
:(?P<second>[0-9][0-9])
|
||
(?:\.(?P<fraction>[0-9]*))?
|
||
(?:[ \t]*(?P<tz>Z|(?P<tz_sign>[-+])(?P<tz_hour>[0-9][0-9]?)
|
||
(?::(?P<tz_minute>[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--<2D>'
|
||
# '\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--<2D>\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 _strip_properties(value):
|
||
# Consume a leading YAML node non-specific tag ('!' + space) and return the
|
||
# remaining node text for normal resolution. PyYAML's SafeLoader applies
|
||
# this TRANSPARENT property and then resolves the following node with its
|
||
# ordinary implicit typing, so (verified against real PyYAML 6.0.3)
|
||
# '! x' -> 'x', '! ' -> null, exactly as the bare node would resolve.
|
||
# Returns _FAIL for every form we do NOT reproduce identically, so the
|
||
# caller fails closed: a tag shorthand ('!x' -> ConstructorError), an
|
||
# explicit or verbatim tag ('!!str' / '!!int' / '!<...>' / '!foo') whose
|
||
# type coercion we do not emulate, a tab separator (PyYAML scanner error),
|
||
# a duplicate tag ('! ! x' -> ParserError).
|
||
#
|
||
# ANCHORS ('&name'): deliberately rejected in FULL, not just duplicates.
|
||
# Real PyYAML tracks anchor NAMES in a document-scoped composer registry
|
||
# and raises ComposerError ("found duplicate anchor") the instant the
|
||
# SAME anchor name is declared on a SECOND node anywhere in the document
|
||
# -- including on a totally unrelated node far from the token field. This
|
||
# line-oriented fallback has no such document-wide registry (and building
|
||
# one reliably, across every node shape this recognizer does not even
|
||
# parse, risks missing a scope and re-opening the fail-open). So instead
|
||
# of emulating the registry, every '&'-anchor property fails closed here,
|
||
# unconditionally. This is a conservative OVER-reject relative to PyYAML:
|
||
# a single, non-duplicated '&a x' is valid YAML that real PyYAML resolves
|
||
# to 'x', but we refuse it too. That is intentional and safe -- fail
|
||
# closed can only cost an emitted token PyYAML would have allowed, never
|
||
# emit one PyYAML would reject.
|
||
seen_tag = False
|
||
while value[:1] in ("!", "&"):
|
||
if value[0] == "!":
|
||
# Non-specific tag: '!' then AT LEAST ONE SPACE. A tab is a PyYAML
|
||
# scanner error; '!x' / '!!type' / '!<verbatim>' are typed/short tags.
|
||
if seen_tag or len(value) < 2 or value[1] != " ":
|
||
return _FAIL
|
||
seen_tag = True
|
||
value = value[1:].lstrip(" ")
|
||
else: # anchor '&name' -- always fail closed, see comment above
|
||
return _FAIL
|
||
return value
|
||
|
||
|
||
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 ("!", "&"):
|
||
# Leading node property (non-specific tag '! ' / plain anchor '&name '):
|
||
# PyYAML applies it to the FOLLOWING node and resolves THAT, so strip the
|
||
# transparent forms and re-resolve the remainder EXACTLY (implicit typing
|
||
# and all). _strip_properties returns _FAIL for the tag/anchor forms
|
||
# PyYAML rejects or type-coerces, which fail closed here.
|
||
remainder = _strip_properties(value)
|
||
if remainder is _FAIL:
|
||
return _FAIL
|
||
return _scalar(remainder)
|
||
if value[0] in _FLOW or value[0] in ("|", ">", "@", "`", '"', "'", "%", ","):
|
||
return _FAIL # flow / block-scalar / reserved / directive / 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
|
||
|
||
|
||
_BREAKS = "\r\n\x85
"
|
||
_VALUE_OPEN_RE = re.compile(r"^\s*(?:-\s+)*[A-Za-z0-9_][A-Za-z0-9_.\-]*:$")
|
||
_SEQ_OPEN_RE = re.compile(r"^\s*(?:-\s+)*-$")
|
||
|
||
|
||
def _value_open_before(prefix):
|
||
# True when a quote appearing at the END of `prefix` (the current logical
|
||
# line so far) begins a flow SCALAR value -- the only position where a quote
|
||
# opens a quoted scalar whose embedded line breaks PyYAML folds instead of
|
||
# splitting. That is: the prefix is empty/all-indent (root or block scalar
|
||
# start), or ends with a mapping ': ' or a sequence '- ' indicator that is
|
||
# SPACE-separated from the quote. A quote glued to the previous char (e.g.
|
||
# 'key:"x') is NOT a value opener (PyYAML needs the space), and a quote mid
|
||
# -content is literal -- both fail this test, so their breaks split the line
|
||
# exactly as before (fail closed / plain-scalar behavior preserved).
|
||
stripped = prefix.rstrip(" ")
|
||
if stripped == "":
|
||
return True # start of line (after any indentation) -> node value start
|
||
if prefix == stripped:
|
||
return False # no space before the quote -> not a value opener
|
||
return bool(_VALUE_OPEN_RE.match(stripped) or _SEQ_OPEN_RE.match(stripped))
|
||
|
||
|
||
def _fold_quoted_break_run(run):
|
||
# Reproduce PyYAML's flow-scalar folding (scan_flow_scalar_spaces +
|
||
# scan_flow_scalar_breaks) for the maximal ' \t' + line-break run inside a
|
||
# quoted scalar. A run with NO break is literal whitespace (verbatim). With a
|
||
# break: surrounding spaces/tabs are dropped; a single \n-class break
|
||
# (\n / \r / \r\n / \x85 NEL) folds to ONE space;
(LS) /
(PS)
|
||
# are kept verbatim; each ADDITIONAL break (a blank line) contributes its own
|
||
# break char (\n for the \n-class). Verified against real PyYAML 6.0.3 for
|
||
# both double- and single-quoted scalars (they fold identically here).
|
||
def _lb(s, j):
|
||
if s[j] == "\r" and j + 1 < len(s) and s[j + 1] == "\n":
|
||
return "\n", 2
|
||
if s[j] in "\r\n\x85":
|
||
return "\n", 1
|
||
return s[j], 1 #
/
preserved verbatim
|
||
n = len(run)
|
||
j = 0
|
||
while j < n and run[j] in " \t":
|
||
j += 1
|
||
if j >= n:
|
||
return run # pure whitespace, no break -> literal content
|
||
line_break, adv = _lb(run, j)
|
||
j += adv
|
||
breaks = []
|
||
while True:
|
||
while j < n and run[j] in " \t":
|
||
j += 1
|
||
if j < n and run[j] in _BREAKS:
|
||
b, adv = _lb(run, j)
|
||
j += adv
|
||
breaks.append(b)
|
||
else:
|
||
break
|
||
out = ""
|
||
if line_break != "\n":
|
||
out += line_break
|
||
elif not breaks:
|
||
out += " "
|
||
return out + "".join(breaks)
|
||
|
||
|
||
def _split_logical_lines(raw_text):
|
||
# Split like str.splitlines() EXCEPT a line break INSIDE a value-opening
|
||
# quoted scalar does NOT end the line: PyYAML keeps the quoted scalar together
|
||
# and folds the break (flow folding above), so we fold it and continue the
|
||
# same logical line. Only the break chars that survive the _FORBIDDEN_CONTROL
|
||
# gate reach here: \n, \r, \x85 (NEL),
(LS),
(PS). OUTSIDE
|
||
# quotes every one ends the line (matching PyYAML's block-level line breaks
|
||
# and str.splitlines), so a NEL/LS/PS used as the document's line-terminator
|
||
# style, and an UNQUOTED plain scalar carrying an embedded break, split and
|
||
# fail closed exactly as before. A quoted scalar left unterminated at EOF
|
||
# stays on the final line and fails closed in _scalar (PyYAML errors too).
|
||
out = []
|
||
cur = []
|
||
i = 0
|
||
n = len(raw_text)
|
||
quote = None
|
||
while i < n:
|
||
ch = raw_text[i]
|
||
if quote is None:
|
||
if ch in _BREAKS:
|
||
out.append("".join(cur))
|
||
cur = []
|
||
i += 2 if (ch == "\r" and i + 1 < n and raw_text[i + 1] == "\n") else 1
|
||
continue
|
||
if ch in ("'", '"') and _value_open_before("".join(cur)):
|
||
quote = ch
|
||
cur.append(ch)
|
||
i += 1
|
||
continue
|
||
cur.append(ch)
|
||
i += 1
|
||
continue
|
||
# inside a value-opening quoted scalar
|
||
if ch == quote:
|
||
cur.append(ch)
|
||
quote = None
|
||
i += 1
|
||
continue
|
||
if ch in " \t" or ch in _BREAKS:
|
||
j = i
|
||
has_break = False
|
||
while j < n and (raw_text[j] in " \t" or raw_text[j] in _BREAKS):
|
||
if raw_text[j] in _BREAKS:
|
||
has_break = True
|
||
j += 1
|
||
run = raw_text[i:j]
|
||
if has_break:
|
||
# A continuation line beginning (at column 0, no indent) with a
|
||
# document marker is a PyYAML scanner error even inside a quoted
|
||
# scalar; when the run ends on a break the continuation content at
|
||
# j is at column 0, so guard it and fail closed. (An INDENTED
|
||
# '---' is literal content to PyYAML and folds normally.)
|
||
if run[-1] in _BREAKS and raw_text[j:j + 3] in ("---", "...") and (
|
||
j + 3 >= n or raw_text[j + 3] in "\0 \t" + _BREAKS
|
||
):
|
||
raise _Bail()
|
||
cur.append(_fold_quoted_break_run(run))
|
||
else:
|
||
cur.append(run)
|
||
i = j
|
||
continue
|
||
cur.append(ch)
|
||
i += 1
|
||
if cur or not out:
|
||
out.append("".join(cur))
|
||
return out
|
||
|
||
|
||
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 line splitting, which would otherwise split on some of these same
|
||
# code points -- e.g. \x0b, \x0c, \x1c-\x1e -- and obscure them) mirrors that
|
||
# exactly: fail closed for the whole document. (NEL 0x85, LS 0x2028 and PS
|
||
# 0x2029 are PRINTABLE to PyYAML's Reader, so they are NOT in
|
||
# _FORBIDDEN_CONTROL; _split_logical_lines below reproduces PyYAML's line-break
|
||
# semantics for them -- a structural break outside quotes, a FOLDED break
|
||
# inside a quoted scalar -- rather than str.splitlines(), which would wrongly
|
||
# split them mid-quote and fail-close a token PyYAML resolves.)
|
||
if _FORBIDDEN_CONTROL.search(raw_text):
|
||
return None
|
||
|
||
lines = _split_logical_lines(raw_text)
|
||
|
||
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
|