241 lines
9.9 KiB
Python
241 lines
9.9 KiB
Python
"""Checks, sets or purges one secret in the machine configs of a Werkator host or in a local env file.
|
|
|
|
Usage: python3 host_secret.py check|set|purge|export KEY [FILE] < {"sha256": "<hex>", "value": "<new value>", "old_sha256": "<hex>"}
|
|
Without FILE, it handles ~/*/.git/werkator/.werkator.yml (lines "KEY: value");
|
|
with FILE, that env file (lines "KEY=value").
|
|
check: tells per file whether the value of KEY has the SHA-256 sha256, and counts the backups that hold that value.
|
|
A backup it cannot read or parse may hold it too: check names it and exits with 2, and purge refuses.
|
|
set: replaces the value of KEY in each file that has KEY, after a backup <file>.bak-secret-<KEY>-<time>,
|
|
and adds KEY to an env file without it.
|
|
It refuses if the files hold more than one old value, or with old_sha256 another one than that,
|
|
and prints the SHA-256 of the old value for purge.
|
|
purge: removes the backups next to each file whose value of KEY has the SHA-256 sha256; never the file itself.
|
|
export: writes {"value": ...} if all files hold one value of KEY, never to a terminal; for tools/rotate-secret fetch.
|
|
Else it prints repository and file names, statuses, counts and the SHA-256 of an old value, never a value.
|
|
Exit code: 0 if all went well, 1 if check found another value, 2 on an error.
|
|
set and purge check every file before they change one.
|
|
"""
|
|
import datetime
|
|
import glob
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
|
|
# Tokens and API keys of Gitea, OpenRouter, TypeSafe and the statistics service consist of these characters;
|
|
# a value with others would need quoting rules in YAML and in the env file.
|
|
VALUE = re.compile(r"[A-Za-z0-9._~+/=-]{16,512}")
|
|
KEY = re.compile(r"[A-Z][A-Z0-9_]*")
|
|
|
|
|
|
class SecretError(Exception):
|
|
pass
|
|
|
|
|
|
def sha256(value):
|
|
return hashlib.sha256(value.encode()).hexdigest()
|
|
|
|
|
|
def _line(key, yaml):
|
|
if yaml:
|
|
return re.compile(rf"^(\s*{key}:[ \t]*)([\"']?)([^\"'\s#]*)\2([ \t]*)$")
|
|
return re.compile(rf"^({key}=)([\"']?)([^\"'\s#]*)\2([ \t]*)$")
|
|
|
|
|
|
def find(lines, key, yaml):
|
|
"""Returns (index, match) of the one line that sets KEY, None without such a line."""
|
|
pattern, named = _line(key, yaml), re.compile(rf"^\s*{key}{':' if yaml else '='}")
|
|
named_lines = [i for i, line in enumerate(lines) if named.match(line)]
|
|
if not named_lines:
|
|
return None
|
|
if len(named_lines) > 1:
|
|
raise SecretError(f"{len(named_lines)} lines set {key}")
|
|
match = pattern.match(lines[named_lines[0]])
|
|
if not match:
|
|
raise SecretError(f"the line of {key} has a form this tool does not change, e.g. a comment")
|
|
return named_lines[0], match
|
|
|
|
|
|
def read_lines(path):
|
|
"""A file it cannot read is an error, not a file without KEY."""
|
|
try:
|
|
with open(path) as f:
|
|
return f.read().split("\n")
|
|
except (OSError, UnicodeDecodeError) as e:
|
|
raise SecretError(f"cannot read {os.path.basename(path)}: {type(e).__name__}") from None
|
|
|
|
|
|
def value_of(path, key, yaml):
|
|
"""The value of KEY in the file, None without KEY."""
|
|
found = find(read_lines(path), key, yaml)
|
|
return found[1].group(3) if found else None
|
|
|
|
|
|
def backups(path):
|
|
"""The backups next to path by the names they have on the hosts, e.g. .werkator.yml.bak-config-<time> or .pre-0.21.0;
|
|
another file with the same prefix, e.g. secrets.env.production, may be a live config."""
|
|
directory, name = os.path.split(path)
|
|
pattern = re.compile(rf"{re.escape(name)}\.(bak|pre|before)([-.][\w.-]*|\d*)")
|
|
return sorted(os.path.join(directory, n) for n in os.listdir(directory or ".")
|
|
if pattern.fullmatch(n) and os.path.isfile(os.path.join(directory, n)))
|
|
|
|
|
|
def holding(path, key, yaml, digest):
|
|
"""Returns (held, unclear): the backups whose value of KEY has the SHA-256 digest,
|
|
and those it cannot read or parse, which may hold it all the same."""
|
|
held, unclear = [], []
|
|
for backup in backups(path):
|
|
try:
|
|
value = value_of(backup, key, yaml)
|
|
except SecretError:
|
|
unclear.append(backup)
|
|
continue
|
|
if value is not None and sha256(value) == digest:
|
|
held.append(backup)
|
|
return held, unclear
|
|
|
|
|
|
def _names(paths):
|
|
return ", ".join(os.path.basename(p) for p in paths)
|
|
|
|
|
|
def write_atomically(path, text):
|
|
"""Werkator may read the file at any time; a rename lets it see the old or the new file, never half of one."""
|
|
directory = os.path.dirname(path) or "."
|
|
mode = os.stat(path).st_mode & 0o777
|
|
fd, temporary = tempfile.mkstemp(dir=directory, prefix=".secret-")
|
|
try:
|
|
os.fchmod(fd, mode)
|
|
with os.fdopen(fd, "w") as f:
|
|
f.write(text)
|
|
os.replace(temporary, path)
|
|
except BaseException:
|
|
if os.path.exists(temporary):
|
|
os.remove(temporary)
|
|
raise
|
|
|
|
|
|
def files(home, env_file):
|
|
if env_file:
|
|
return [(os.path.basename(env_file), env_file, False)]
|
|
paths = sorted(glob.glob(os.path.join(home, "*", ".git", "werkator", ".werkator.yml")))
|
|
return [(os.path.relpath(p, home).split(os.sep)[0], p, True) for p in paths]
|
|
|
|
|
|
def check(targets, key, digest):
|
|
worst = 0
|
|
for name, path, yaml in targets:
|
|
value = value_of(path, key, yaml)
|
|
if value is None:
|
|
continue
|
|
held, unclear = holding(path, key, yaml, digest)
|
|
status = "equal" if sha256(value) == digest else "differs"
|
|
print(f"{name}: {status}; {len(held)} backups hold that value"
|
|
+ (f"; cannot read or parse {_names(unclear)}" if unclear else ""))
|
|
worst = max(worst, 2 if unclear else 0 if status == "equal" else 1)
|
|
return worst
|
|
|
|
|
|
def set_value(targets, key, value, old_digest, now):
|
|
if not VALUE.fullmatch(value):
|
|
raise SecretError("the new value is empty, too long or has characters other than A-Z a-z 0-9 . _ ~ + / = -")
|
|
planned = []
|
|
for name, path, yaml in targets:
|
|
lines = read_lines(path)
|
|
found = find(lines, key, yaml)
|
|
if found or not yaml:
|
|
planned.append((name, path, yaml, lines, found))
|
|
# A file with a third value would mean that an earlier rotation stopped half-way; the human sorts it out first.
|
|
old = {sha256(found[1].group(3)) for *_, found in planned if found and found[1].group(3) != value}
|
|
if old_digest and old - {old_digest}:
|
|
raise SecretError(f"a file holds another value of {key} than the old one")
|
|
if len(old) > 1:
|
|
raise SecretError(f"the files hold {len(old)} different old values of {key}")
|
|
if not planned:
|
|
raise SecretError(f"no file sets {key}")
|
|
for name, path, yaml, lines, found in planned:
|
|
if found and found[1].group(3) == value:
|
|
print(f"{name}: equal")
|
|
continue
|
|
backup = f"{path}.bak-secret-{key}-{now}"
|
|
with open(path) as f, open(os.open(backup, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as b:
|
|
b.write(f.read())
|
|
if found:
|
|
index, match = found
|
|
quote = match.group(2)
|
|
lines[index] = f"{match.group(1)}{quote}{value}{quote}{match.group(4)}"
|
|
else:
|
|
lines.insert(len(lines) - 1 if lines[-1] == "" else len(lines), f"{key}={value}")
|
|
write_atomically(path, "\n".join(lines))
|
|
print(f"{name}: {'replaced' if found else 'added'}; backup {os.path.basename(backup)}")
|
|
if old:
|
|
print(f"old value: sha256 {old.pop()}")
|
|
return 0
|
|
|
|
|
|
def purge(targets, key, digest):
|
|
for name, path, yaml in targets:
|
|
if not os.path.exists(path):
|
|
continue
|
|
current = value_of(path, key, yaml)
|
|
if current is not None and sha256(current) == digest:
|
|
raise SecretError(f"{name} still holds that value; set the new value first")
|
|
unclear = holding(path, key, yaml, digest)[1]
|
|
if unclear:
|
|
raise SecretError(f"{name}: cannot read or parse {_names(unclear)}; check or remove them by hand")
|
|
for name, path, yaml in targets:
|
|
if not os.path.exists(path):
|
|
continue
|
|
removed = holding(path, key, yaml, digest)[0]
|
|
for backup in removed:
|
|
os.remove(backup)
|
|
if removed:
|
|
print(f"{name}: removed {_names(removed)}")
|
|
return 0
|
|
|
|
|
|
def export(targets, key, stdout):
|
|
"""Writes the one value of KEY as JSON, for tools/rotate-secret fetch; never to a terminal."""
|
|
if stdout.isatty():
|
|
raise SecretError("export writes a value; it refuses a terminal")
|
|
values = {value_of(path, key, yaml) for _, path, yaml in targets} - {None}
|
|
if len(values) != 1:
|
|
raise SecretError(f"the files hold {len(values)} different values of {key}")
|
|
stdout.write(json.dumps({"value": values.pop()}))
|
|
return 0
|
|
|
|
|
|
def _digest(value):
|
|
if value is not None and not re.fullmatch(r"[0-9a-f]{64}", value):
|
|
raise SecretError("a SHA-256 must be 64 lower-case hex digits")
|
|
return value
|
|
|
|
|
|
def main(argv, stdin, home, now=None):
|
|
if len(argv) not in (2, 3) or argv[0] not in ("check", "set", "purge", "export") or not KEY.fullmatch(argv[1]):
|
|
print(__doc__.strip(), file=sys.stderr)
|
|
return 2
|
|
mode, key = argv[0], argv[1]
|
|
targets = files(home, argv[2] if len(argv) == 3 else None)
|
|
now = now or datetime.datetime.now().strftime("%Y%m%d%H%M%S")
|
|
try:
|
|
if mode == "export":
|
|
return export(targets, key, sys.stdout)
|
|
payload = json.load(stdin)
|
|
if mode == "set":
|
|
return set_value(targets, key, payload["value"], _digest(payload.get("old_sha256")), now)
|
|
digest = _digest(payload["sha256"])
|
|
return check(targets, key, digest) if mode == "check" else purge(targets, key, digest)
|
|
except (SecretError, OSError) as e:
|
|
print(f"error: {e}", file=sys.stderr)
|
|
except (KeyError, TypeError, ValueError):
|
|
# A JSONDecodeError would quote the payload, and the payload holds the value.
|
|
print("error: the input is no JSON object with the needed field", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:], sys.stdin, os.path.expanduser("~")))
|