fix(git-wrappers): #865 round-7 addendum — conservative YAML recognizer + review/comment hardening
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
All checks were successful
ci/woodpecker/pr/ci Pipeline was successful
Fold the remaining round-7 audit items into the tea-CLI comment-invocation fix. Fallback token parser (detect-platform.sh, test-gitea-login-resolution.sh): Replace the line-by-line scalar fallback with a strict CONSERVATIVE block-YAML recognizer that reconstructs the same object PyYAML would or fails closed the instant it meets anything outside the tea-config subset. Closes 5 structural fail-open classes the old parser missed (nested-shadow logins, block-scalar shadow, duplicate root key / login name / token field, malformed-after-valid, extra-document / end-marker). Validated by a 360k-check differential fuzz vs real PyYAML (0 fail-open) plus explicit forced-PyYAML-absence fixtures. ITEM 1 (pr-review.sh) current-head TOCTOU: after the exact review-id read-back succeeds, re-read the live PR head and fail closed if it advanced past the submitted commit_id, so a review is never reported as covering a superseded tip. ITEM 2 (pr-review.sh comment action): require the returned resource be a pull_request (populated pull_request_url); reject a bare issue_url so a plain issue #N cannot masquerade as a verified PR comment. issue-comment.sh keeps its broader issue-or-PR acceptance. ITEM 3a (both wrappers): move the Authorization bearer OUT of curl argv into a private mode-0600 curl --config file (gitea_write_auth_config), removed on every exit path, so the token never appears in the process table. ITEM 3b (pr-review.sh): bind the review body with presence + string-type + exact equality instead of `(body or "")`, so a non-empty submitted body persisted as null/missing fails closed. Tests: add race, plain-issue, argv-capture (no token printed), and null-body fixtures; broaden temp-leak checks to the auth-config files. Full gate set green (bash -n, shellcheck -x -S warning, prettier, all 3 REST/resolution suites with PyYAML and forced-absent, cold TURBO_FORCE turbo 14/14). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -563,6 +563,30 @@ get_gitea_token() {
|
||||
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
|
||||
@@ -647,46 +671,255 @@ def _implicit_nonstring(text):
|
||||
)
|
||||
|
||||
|
||||
def _strip_scalar(value):
|
||||
# Resolve a YAML flow scalar the way PyYAML would for tea's simple scalars:
|
||||
# honor surrounding quotes and strip a trailing inline comment. A quoted
|
||||
# scalar keeps its literal contents (any '#' inside is data, not a comment);
|
||||
# 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".
|
||||
# Sentinel: "outside the supported subset -> fail closed". Distinct from a
|
||||
# genuine null (None), which is a valid resolved value.
|
||||
_FAIL = object()
|
||||
_KEY_RE = re.compile(r"^([A-Za-z0-9_][A-Za-z0-9_.\-]*)[ \t]*:(?:[ \t](.*)|)$")
|
||||
_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, so its contents are returned verbatim.
|
||||
# An UNQUOTED scalar, however, is subject to PyYAML's implicit typing: forms
|
||||
# like `12345`, `null`, `~`, `yes`/`true`, `3.14` or a timestamp resolve to a
|
||||
# non-string (int/None/bool/float/date), which PyYAML's path would reject as
|
||||
# a token via _accept's isinstance(str) guard. To stay faithful (and never
|
||||
# fail OPEN by surfacing a stringified non-string as a credential) such a
|
||||
# scalar returns None here so the caller treats it as absent (fail closed),
|
||||
# exactly as the PyYAML path does. Only unquoted scalars PyYAML would type as
|
||||
# a string are returned as a string.
|
||||
value = value.strip()
|
||||
# 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).
|
||||
value = raw.strip()
|
||||
if not value:
|
||||
return value
|
||||
return None # empty plain scalar -> null
|
||||
if value[0] in ("'", '"'):
|
||||
quote = value[0]
|
||||
end = value.find(quote, 1)
|
||||
if end != -1:
|
||||
return value[1:end]
|
||||
# Unterminated quote: PyYAML would error; return best-effort remainder so
|
||||
# the (more conservative) caller still compares against the wanted name.
|
||||
return value[1:]
|
||||
for index, char in enumerate(value):
|
||||
if char == "#" and (index == 0 or value[index - 1] in (" ", "\t")):
|
||||
value = value[:index]
|
||||
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()
|
||||
if not value or _implicit_nonstring(value):
|
||||
# Unquoted scalar that PyYAML resolves to null or another non-string
|
||||
# type: fail closed rather than emit a stringified non-credential.
|
||||
if not value:
|
||||
return None
|
||||
if value[0] in _FLOW or value[0] in ("|", ">", "?", "@", "`", '"', "'"):
|
||||
return _FAIL # flow / block-scalar / reserved / anchor / quote indicator
|
||||
if any(ch in _FLOW for ch in value):
|
||||
return _FAIL
|
||||
if ": " in value or value.endswith(":") or "\t" in value:
|
||||
return _FAIL # nested-mapping-in-scalar / ambiguous
|
||||
if _implicit_nonstring(value):
|
||||
return None # non-string implicit type -> null-equivalent for a token
|
||||
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
|
||||
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()
|
||||
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
|
||||
@@ -741,81 +974,27 @@ def _token_via_pyyaml():
|
||||
|
||||
|
||||
def _token_via_lines():
|
||||
# Conservative fallback for hosts without PyYAML. tea writes config.yml in a
|
||||
# fixed, flat shape (a `logins:` list of maps with scalar name/url/token
|
||||
# fields). This scan is SCOPE-AWARE: a field is attributed to a login entry
|
||||
# ONLY when it sits at that entry's own direct-field indentation. A field
|
||||
# nested inside a deeper sub-map (e.g. `extra:\n token: X`) or a mis-indented
|
||||
# line is NEVER attached to the entry — exactly the cases where PyYAML resolves
|
||||
# the entry's own `token` to None (or errors) and thus fails closed. Likewise
|
||||
# only list items at the login list's own dash indent open a new entry, so a
|
||||
# nested list item cannot masquerade as a sibling login. It returns the
|
||||
# `token` of the entry whose `name` EXACTLY equals the requested login AND
|
||||
# whose url host/port matches the repo host; anything else yields None (fail
|
||||
# closed). This can only ever be MORE conservative than PyYAML (it never
|
||||
# selects a token where PyYAML would refuse), never less.
|
||||
# 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:
|
||||
lines = handle.read().splitlines()
|
||||
|
||||
logins_indent = None
|
||||
start = len(lines)
|
||||
for index, line in enumerate(lines):
|
||||
match = re.match(r"^(\s*)logins\s*:\s*$", line)
|
||||
if match:
|
||||
logins_indent = len(match.group(1))
|
||||
start = index + 1
|
||||
break
|
||||
if logins_indent is None:
|
||||
config = _safe_parse(lines)
|
||||
if config is _FAIL:
|
||||
return None
|
||||
|
||||
entries = []
|
||||
current = None
|
||||
item_indent = None # dash column of the login list's own items
|
||||
field_indent = None # exact column of the current entry's direct fields
|
||||
for line in lines[start:]:
|
||||
if not line.strip() or line.lstrip().startswith("#"):
|
||||
continue
|
||||
indent = len(line) - len(line.lstrip(" "))
|
||||
if indent <= logins_indent:
|
||||
break
|
||||
item = re.match(r"^(\s*)-(\s*)(.*)$", line)
|
||||
if item:
|
||||
dash_indent = len(item.group(1))
|
||||
if item_indent is None:
|
||||
item_indent = dash_indent
|
||||
if dash_indent != item_indent:
|
||||
# A more-deeply-indented (nested) or dedented list item: not a
|
||||
# direct login entry. Ignore it and its scope.
|
||||
continue
|
||||
current = {}
|
||||
entries.append(current)
|
||||
content = item.group(3)
|
||||
if content:
|
||||
# First field shares this line; its column is the direct-field
|
||||
# indent for the rest of the entry.
|
||||
field_indent = dash_indent + 1 + len(item.group(2))
|
||||
pair = re.match(r"^([A-Za-z0-9_]+)\s*:\s*(.*)$", content)
|
||||
if pair:
|
||||
current[pair.group(1)] = _strip_scalar(pair.group(2))
|
||||
else:
|
||||
# Bare "-": the first following field line establishes the indent.
|
||||
field_indent = None
|
||||
continue
|
||||
if current is None:
|
||||
continue
|
||||
if field_indent is None:
|
||||
field_indent = indent
|
||||
if indent != field_indent:
|
||||
# Deeper => a nested sub-map's field (not this entry's own); anything
|
||||
# else at an unexpected column is not a direct field. Skip either way.
|
||||
continue
|
||||
pair = re.match(r"^([A-Za-z0-9_]+)\s*:\s*(.*)$", line.strip())
|
||||
if pair:
|
||||
current[pair.group(1)] = _strip_scalar(pair.group(2))
|
||||
|
||||
for entry in entries:
|
||||
if str(entry.get("name") or "") == wanted:
|
||||
return _accept(entry.get("token"), entry.get("url"))
|
||||
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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user