tegwick
6ae4708ea0
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a12d13-27f2-7163-b8c0-f682af02772c
83 lines
3.3 KiB
Python
83 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate and atomically update a private generated credential file.
|
|
|
|
The caller supplies the new value on stdin, never argv. This protects the
|
|
handoff/output boundary; it does not isolate a hostile process with the same UID.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
import stat
|
|
import sys
|
|
|
|
|
|
class Refused(Exception):
|
|
pass
|
|
|
|
|
|
def private(info, directory=False):
|
|
kind = stat.S_ISDIR if directory else stat.S_ISREG
|
|
if not kind(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
|
|
raise Refused()
|
|
if not directory and info.st_nlink != 1:
|
|
raise Refused()
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 5 or sys.argv[1] not in ('check', 'update'):
|
|
raise Refused()
|
|
action, root_arg, component, key = sys.argv[1:]
|
|
if component not in ('privacyidea', 'lldap', 'authelia', 'breakglass'):
|
|
raise Refused()
|
|
if not re.fullmatch(r'[A-Z][A-Z0-9_]+', key):
|
|
raise Refused()
|
|
root = Path(os.path.abspath(root_arg))
|
|
# Refuse symlink ancestors rather than resolving them silently.
|
|
for part in (root, *root.parents):
|
|
if part.is_symlink(): raise Refused()
|
|
root_fd = parent_fd = file_fd = None
|
|
temporary = None
|
|
try:
|
|
flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
|
|
root_fd = os.open(root, flags)
|
|
private(os.fstat(root_fd), directory=True)
|
|
parent_fd = os.open(component, flags, dir_fd=root_fd)
|
|
private(os.fstat(parent_fd), directory=True)
|
|
file_fd = os.open('secrets.env', os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent_fd)
|
|
original = os.fstat(file_fd)
|
|
private(original)
|
|
raw = os.read(file_fd, 65537)
|
|
if len(raw) > 65536: raise Refused()
|
|
lines = raw.decode('utf-8').splitlines(keepends=True)
|
|
matches = [i for i, line in enumerate(lines) if line.startswith(key + '=')]
|
|
if len(matches) != 1: raise Refused()
|
|
if action == 'check': return
|
|
value = sys.stdin.buffer.read(257).decode('ascii').strip()
|
|
if not re.fullmatch(r'[A-Za-z0-9]{32,128}', value): raise Refused()
|
|
lines[matches[0]] = key + '=' + value + '\n'
|
|
temporary = '.rotation-' + secrets.token_hex(12)
|
|
output_fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
|
|
0o600, dir_fd=parent_fd)
|
|
with os.fdopen(output_fd, 'w', encoding='utf-8') as output:
|
|
output.writelines(lines); output.flush(); os.fsync(output.fileno())
|
|
current = os.stat('secrets.env', dir_fd=parent_fd, follow_symlinks=False)
|
|
if (current.st_dev, current.st_ino, current.st_mtime_ns, current.st_size) != (
|
|
original.st_dev, original.st_ino, original.st_mtime_ns, original.st_size):
|
|
raise Refused()
|
|
os.replace(temporary, 'secrets.env', src_dir_fd=parent_fd, dst_dir_fd=parent_fd)
|
|
temporary = None
|
|
os.fsync(parent_fd)
|
|
finally:
|
|
if temporary is not None and parent_fd is not None:
|
|
os.unlink(temporary, dir_fd=parent_fd)
|
|
for fd in (file_fd, parent_fd, root_fd):
|
|
if fd is not None: os.close(fd)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
main()
|
|
except (Refused, OSError, ValueError, UnicodeError):
|
|
print('ERROR: private credential file validation/update failed; no values are logged.', file=sys.stderr)
|
|
sys.exit(1)
|