acf7955f87
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
Round-9 auditor blockers, both fixed at root cause in the PyYAML-absent
line-parser fallback of get_gitea_token_for_login:
1. Control-character fail-open (dangerous): PyYAML 6.0.3's Reader rejects
the WHOLE document (ReaderError) if a forbidden C0/DEL control byte
{0x00-0x08, 0x0B, 0x0C, 0x0E-0x1F, 0x7F} appears ANYWHERE in the raw
stream -- plain scalar, inside quotes, or a comment -- regardless of
position, verified empirically against the real installed PyYAML 6.0.3.
The fallback previously only guarded tabs and emitted the login token
from such documents. Added a whole-document _FORBIDDEN_CONTROL scan on
the raw text (before splitlines(), which itself splits on some of the
same bytes) so the fallback fails closed identically to PyYAML.
2. Unsigned-exponent float over-rejection: PyYAML's implicit float
resolver requires an EXPLICIT sign on the exponent; unsigned-exponent
spellings (1.0e10, +1.0e10, -1.0e10, 1.0E10, .5e10, 4.e8) are PyYAML
STRINGS, not floats. The fallback's _IMPLICIT_FLOAT pattern allowed an
optional sign, misclassifying these as floats and dropping the token.
Tightened the regex to match PyYAML's resolver exactly.
Also folds in a residual over-rejection found via this round's wide
differential fuzz (1767 cases, 0 fail-opens / 0 over-rejections after the
fix): a plain scalar starting with "?" not followed by whitespace (e.g.
"?x") is a valid PyYAML string, but the fallback's blanket-reject set
treated every leading "?" as illegal. Narrowed to match the existing
space-aware handling already used for "-" and ":".
Adds regression fixtures to test-gitea-login-resolution.sh covering all
three fixes under both PyYAML-present and forced-ImportError-absent runs.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1260 lines
46 KiB
Bash
Executable File
1260 lines
46 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 bytes 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). Empirically verified against real PyYAML 6.0.3: EVERY C0
|
|
# control byte below plus DEL (0x7F) rejects in plain scalars, inside
|
|
# double-quoted scalars, and inside single-quoted scalars alike; only
|
|
# TAB(0x09), LF(0x0A), and CR(0x0D) are accepted among the C0 range (LF/CR are
|
|
# line separators, handled elsewhere; TAB has its own narrower, position-aware
|
|
# guard above since PyYAML accepts it in some contexts). This set therefore
|
|
# excludes 0x09/0x0A/0x0D and matches the forbidden C0 set {0x00-0x08, 0x0B,
|
|
# 0x0C, 0x0E-0x1F} plus DEL (0x7F).
|
|
_FORBIDDEN_CONTROL = re.compile(
|
|
"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]"
|
|
)
|
|
|
|
# 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
|
|
if any(ch in _FLOW for ch in value):
|
|
return _FAIL
|
|
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 forbidden C0/DEL control byte 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 control bytes -- e.g. \x0b, \x0c, \x1c-\x1e
|
|
# -- and would otherwise obscure them) mirrors that exactly: fail closed
|
|
# for the whole document rather than risk emitting a token PyYAML would
|
|
# have refused.
|
|
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
|