139 lines
6.9 KiB
Python
139 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Privileged READ ONLY k3s selector preflight; emit fixed predicates, never values."""
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
|
|
UNIT_SHA = 'b5398c385430ce225faadc1f3724597fe4b05dd999a04b8abbca64d4b0d68c0e'
|
|
MACHINE_SHA = 'fe7367a60951922122af7ba18ed2cde65c7e0a232bbedfb045cb575de2094c2f'
|
|
ENV_FILES = ('/etc/default/k3s', '/etc/sysconfig/k3s', '/etc/systemd/system/k3s.service.env')
|
|
SELECTORS = ('K3S_CONFIG_FILE', 'K3S_KUBECONFIG_MODE', 'K3S_KUBECONFIG_OUTPUT', 'K3S_KUBECONFIG_GROUP')
|
|
|
|
|
|
def read_fixed(path, max_bytes=1024*1024):
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
|
|
try:
|
|
before = os.fstat(fd)
|
|
if not stat.S_ISREG(before.st_mode) or before.st_uid != 0 or before.st_mode & 0o022:
|
|
raise ValueError('file_custody')
|
|
with os.fdopen(os.dup(fd), 'rb') as stream:
|
|
data = stream.read(max_bytes+1)
|
|
after = os.fstat(fd)
|
|
if len(data) > max_bytes or (before.st_size,before.st_mtime_ns,before.st_ctime_ns) != (after.st_size,after.st_mtime_ns,after.st_ctime_ns):
|
|
raise ValueError('file_changed_or_oversized')
|
|
return data, before
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def env_predicates(raw, *, process=False):
|
|
"""Only key presence leaves this function. No shell evaluation or raw errors."""
|
|
found = set()
|
|
valid = True
|
|
for line in (raw.split(b'\0') if process else raw.splitlines()):
|
|
line = line.strip()
|
|
if not line or (not process and line.startswith((b'#',b';'))):
|
|
continue
|
|
match = re.fullmatch(rb'([A-Za-z_][A-Za-z_0-9]*)=(.*)',line)
|
|
if not match:
|
|
valid = False
|
|
continue
|
|
name = match.group(1).decode('ascii')
|
|
if name in found:
|
|
valid = False
|
|
found.add(name)
|
|
if not process:
|
|
# Conservative subset of systemd EnvironmentFile, not a shell parser.
|
|
# Complex/multiline/escape forms require owner review, never guess.
|
|
value = match.group(2)
|
|
if b'\\' in value or value.count(b'"') not in (0,2) or value.count(b"'") not in (0,2):
|
|
valid = False
|
|
return {'syntax_supported': valid,
|
|
'selectors_absent': all(key not in found for key in SELECTORS),
|
|
'selector_presence': {key:key in found for key in SELECTORS}}
|
|
|
|
|
|
def show(prop):
|
|
result = subprocess.run(['/usr/bin/systemctl','show','k3s','--property='+prop,'--value'],
|
|
capture_output=True,timeout=10,check=False)
|
|
if result.returncode:
|
|
raise ValueError('systemd_read_failed')
|
|
return result.stdout.strip()
|
|
|
|
|
|
def inspect():
|
|
if os.geteuid() != 0:
|
|
return {'status':'refused_requires_attended_root','read_only':True,'passed':False}
|
|
checks = {}
|
|
checks['expected_machine'] = hashlib.sha256(Path('/etc/machine-id').read_bytes()).hexdigest() == MACHINE_SHA
|
|
if not checks['expected_machine']:
|
|
return {'status':'refused_target','read_only':True,'passed':False,'checks':checks}
|
|
unit, meta = read_fixed('/etc/systemd/system/k3s.service')
|
|
checks['pinned_unit'] = hashlib.sha256(unit).hexdigest() == UNIT_SHA
|
|
checks['unit_identity'] = (meta.st_dev,meta.st_ino,meta.st_gid,stat.S_IMODE(meta.st_mode),meta.st_nlink)==(2049,7875,0,0o644,1)
|
|
if not all(checks.values()):
|
|
return {'status':'refused_unit_drift','read_only':True,'passed':False,'checks':checks}
|
|
checks['fragment_expected'] = show('FragmentPath') == b'/etc/systemd/system/k3s.service'
|
|
checks['no_dropins'] = show('DropInPaths') == b''
|
|
checks['no_pending_reload'] = show('NeedDaemonReload') == b'no'
|
|
checks['active'] = show('ActiveState') == b'active' and show('SubState') == b'running'
|
|
pid = show('MainPID')
|
|
if not re.fullmatch(rb'[1-9][0-9]*',pid):
|
|
raise ValueError('main_pid_unavailable')
|
|
proc = Path('/proc')/pid.decode('ascii')
|
|
proc_stat = (proc/'stat').read_bytes()
|
|
start = proc_stat.rsplit(b')',1)[1].split()[19]
|
|
# Read only this bound process environment. Never emit raw bytes or digest.
|
|
environment = (proc/'environ').read_bytes()
|
|
effective = env_predicates(environment,process=True)
|
|
checks['effective_env_simple'] = effective['syntax_supported']
|
|
checks['effective_selectors_absent'] = effective['selectors_absent']
|
|
argv = (proc/'cmdline').read_bytes().split(b'\0')
|
|
checks['effective_mode644'] = argv.count(b'--write-kubeconfig-mode=644') == 1
|
|
checks['no_cli_config_selector'] = not any(x in (b'-c',b'--config') or x.startswith(b'--config=') or (x.startswith(b'-c') and len(x)>2) for x in argv)
|
|
refs=[]
|
|
for match in re.finditer(rb'^EnvironmentFile=([^\r\n]+)',unit,re.M):
|
|
value=match.group(1).strip();optional=value.startswith(b'-');value=value.lstrip(b'-')
|
|
if value.decode('utf-8') not in ENV_FILES:
|
|
raise ValueError('unknown_environment_source')
|
|
refs.append((value.decode('utf-8'),optional))
|
|
checks['known_environment_sources'] = len(refs)>0 and len({x[0] for x in refs})==len(refs)
|
|
file_results={}
|
|
for path,optional in refs:
|
|
try:
|
|
raw, st = read_fixed(path)
|
|
predicates=env_predicates(raw)
|
|
file_results[path]={'present':True,**predicates}
|
|
except FileNotFoundError:
|
|
file_results[path]={'present':False,'syntax_supported':optional,'selectors_absent':optional}
|
|
checks['environment_file_'+str(ENV_FILES.index(path))] = file_results[path]['syntax_supported'] and file_results[path]['selectors_absent']
|
|
checks['default_config_absent'] = not Path('/etc/rancher/k3s/config.yaml').exists()
|
|
directory=Path('/etc/rancher/k3s/config.yaml.d')
|
|
checks['default_dropins_absent'] = not directory.exists() or (directory.is_dir() and not directory.is_symlink() and not any(directory.iterdir()))
|
|
# Revalidate identity to refuse PID reuse/config replacement during inspection.
|
|
checks['same_process'] = show('MainPID') == pid and (proc/'stat').read_bytes().rsplit(b')',1)[1].split()[19] == start
|
|
checks['same_unit'] = hashlib.sha256(read_fixed('/etc/systemd/system/k3s.service')[0]).hexdigest()==UNIT_SHA
|
|
return {'status':'preflight_passed' if all(checks.values()) else 'owner_review_required',
|
|
'passed':all(checks.values()),'read_only':True,'checks':checks,
|
|
'environment_files':file_results,'effective_selector_presence':effective['selector_presence'],
|
|
'target':'railiance01','captured_at_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
'protected_values_emitted':False,'unit_change_authorized':False,'assignment_admitted':False}
|
|
|
|
|
|
def main():
|
|
try:
|
|
receipt=inspect()
|
|
except BaseException:
|
|
# No exception text/traceback, even when a parser embeds protected input.
|
|
receipt={'status':'preflight_refused','passed':False,'read_only':True,'protected_values_emitted':False}
|
|
print(json.dumps(receipt,sort_keys=True))
|
|
return 0 if receipt['passed'] else 1
|
|
|
|
|
|
if __name__=='__main__':
|
|
raise SystemExit(main())
|