40 lines
1.6 KiB
Python
40 lines
1.6 KiB
Python
"""Pure reviewed k3s unit transformation. No write, reload or restart capability."""
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
|
|
BEFORE = 'b5398c385430ce225faadc1f3724597fe4b05dd999a04b8abbca64d4b0d68c0e'
|
|
AFTER = '866ea58bfab00215e5bd3352287ebb0230d4ddb7dd3ed4b1cd853e74c6698d7d'
|
|
PATH = Path('/etc/systemd/system/k3s.service')
|
|
|
|
|
|
def transform(raw, *, before=BEFORE, after=AFTER):
|
|
if hashlib.sha256(raw).hexdigest() != before:
|
|
raise ValueError('unit_preimage_changed')
|
|
candidate, count = re.subn(rb'(--write-kubeconfig-mode(?:=|[ \t]+)[\x22\x27]?)644\b', rb'\g<1>600', raw)
|
|
if count != 1 or len(candidate) != len(raw) or sum(a != b for a,b in zip(raw,candidate)) != 2:
|
|
raise ValueError('unit_change_not_exact')
|
|
if hashlib.sha256(candidate).hexdigest() != after:
|
|
raise ValueError('unit_candidate_mismatch')
|
|
return candidate
|
|
|
|
|
|
def prepare():
|
|
# Open without symlink following and bind metadata/read to the same object.
|
|
fd = os.open(PATH, os.O_RDONLY | os.O_NOFOLLOW)
|
|
try:
|
|
st = os.fstat(fd)
|
|
if (not stat.S_ISREG(st.st_mode) or stat.S_IMODE(st.st_mode) != 0o644
|
|
or (st.st_uid, st.st_gid, st.st_dev, st.st_ino) != (0,0,2049,7875)
|
|
or st.st_nlink != 1):
|
|
raise ValueError('unit_identity_changed')
|
|
with os.fdopen(os.dup(fd), 'rb') as stream:
|
|
candidate = transform(stream.read())
|
|
end = os.fstat(fd)
|
|
if (end.st_mtime_ns,end.st_ctime_ns,end.st_size) != (st.st_mtime_ns,st.st_ctime_ns,st.st_size):
|
|
raise ValueError('unit_changed_during_read')
|
|
return candidate
|
|
finally:
|
|
os.close(fd)
|