208 lines
10 KiB
Python
208 lines
10 KiB
Python
#!/usr/bin/python3
|
|
"""Durable FD-pinned public OS staging on an existing protected data inode.
|
|
|
|
Root-owned coordinator state stays on the OS filesystem. No directory is created
|
|
through the app-owned data ancestry. Mount recovery authenticates original and
|
|
alias inodes; a pathname swap, unknown partial object or foreign mount refuses.
|
|
"""
|
|
import argparse
|
|
import ctypes
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import stat
|
|
import time
|
|
|
|
from d09_store import atomic, read, secure
|
|
import os_update_custody as custody
|
|
|
|
LIBC=ctypes.CDLL(None,use_errno=True)
|
|
MS_BIND=4096
|
|
MS_PRIVATE=1<<18
|
|
CONFIG=Path('/etc/archipelago-os-custody.json')
|
|
STATE=Path('/var/lib/archipelago-delivery-coordinator')
|
|
|
|
def require(ok,reason):
|
|
if not ok:raise ValueError(reason)
|
|
|
|
def identity(fd):
|
|
m=os.fstat(fd)
|
|
return [m.st_dev,m.st_ino,m.st_uid,m.st_gid,stat.S_IMODE(m.st_mode)]
|
|
|
|
def pin_source(path,expected):
|
|
path=Path(path)
|
|
require(path.is_absolute() and '..' not in path.parts,'canonical original support path required')
|
|
fd=os.open('/',os.O_PATH|os.O_DIRECTORY|os.O_CLOEXEC)
|
|
try:
|
|
for component in path.parts[1:]:
|
|
child=os.open(component,os.O_PATH|os.O_DIRECTORY|os.O_NOFOLLOW|os.O_CLOEXEC,dir_fd=fd)
|
|
os.close(fd);fd=child
|
|
require(identity(fd)==expected and expected[2:4]==[0,0] and not expected[4]&0o022,
|
|
'original support inode/custody differs')
|
|
return fd
|
|
except BaseException:os.close(fd);raise
|
|
|
|
def mounted(path):
|
|
# All production destination paths are fixed ASCII without escaped fields.
|
|
return sum(row.split()[4]==str(path) for row in Path('/proc/self/mountinfo').read_text().splitlines())
|
|
|
|
class Coordinator:
|
|
def __init__(self,config=CONFIG,state=STATE):
|
|
require(os.getresuid()==(0,0,0) and os.getresgid()==(0,0,0),'root custody coordinator required')
|
|
self.config=Path(config);self.state=Path(state);self.pins=[];self.lock=None;self.state_pin=None
|
|
self.raw=read(self.config);self.digest=hashlib.sha256(self.raw).hexdigest()
|
|
self.value=json.loads(self.raw)
|
|
v=self.value
|
|
require(type(v) is dict and set(v)=={'version','node','source','source_identity','alias','os_root','stage_name'},
|
|
'OS custody appointment fields')
|
|
require(type(v['version']) is int and v['version']==1 and v['node']==os.uname().nodename and v['stage_name']=='delivery-os-custody',
|
|
'OS custody node/generation binding')
|
|
require(v['source']=='/var/lib/archipelago/support' and
|
|
v['alias']=='/var/lib/archipelago-delivery-support' and
|
|
v['os_root']=='/var/lib/archipelago-os-update','fixed OS custody locators required')
|
|
expected=v['source_identity']
|
|
require(type(expected) is list and len(expected)==5 and all(type(n) is int and n>=0 for n in expected),
|
|
'source identity schema')
|
|
self.journal=self.state/'mounts.json'
|
|
try:
|
|
self.state_pin,_=custody.mkdir(self.state)
|
|
self.lock=os.open(self.state/'owner.lock',os.O_RDWR|os.O_CREAT|os.O_NOFOLLOW|os.O_CLOEXEC,0o600)
|
|
m=os.fstat(self.lock)
|
|
require(stat.S_ISREG(m.st_mode) and m.st_uid==0 and m.st_nlink==1 and stat.S_IMODE(m.st_mode)==0o600,
|
|
'coordinator lock custody')
|
|
fcntl.flock(self.lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
|
|
self.lock_identity=identity(self.lock)
|
|
if self.journal.exists():
|
|
self.head=json.loads(read(self.journal))
|
|
require(set(self.head)=={'version','configuration_sha256','sequence','phase','objects'} and
|
|
type(self.head['version']) is int and self.head['version']==1 and self.head['configuration_sha256']==self.digest and
|
|
type(self.head['sequence']) is int and self.head['sequence']>=0,
|
|
'durable custody journal binding')
|
|
require(type(self.head['objects']) is dict and set(self.head['objects'])<=
|
|
{'support-mountpoint','stage','os-mountpoint'},'durable custody object inventory')
|
|
for name,value in self.head['objects'].items():
|
|
require(type(value) is list and len(value)==5 and all(type(n) is int and n>=0 for n in value) and
|
|
value[2:]==[0,0,0o700],'durable root-private object identity')
|
|
else:
|
|
require(set(os.listdir(self.state_pin.fd))=={'owner.lock'},'unreconciled coordinator state')
|
|
self.head={'version':1,'configuration_sha256':self.digest,'sequence':0,'phase':'appointed','objects':{}}
|
|
self.commit('appointed')
|
|
except BaseException:self.close();raise
|
|
|
|
def commit(self,phase):
|
|
self.state_pin.check()
|
|
self.head['sequence']+=1;self.head['phase']=phase
|
|
atomic(self.journal,json.dumps(self.head,sort_keys=True,separators=(',',':')).encode())
|
|
self.state_pin.check()
|
|
|
|
def directory(self,name,path):
|
|
objects=self.head['objects']
|
|
if name not in objects:
|
|
self.commit('create-'+name+'-intent')
|
|
pin,created=custody.mkdir(path)
|
|
try:
|
|
require(created,'unrecorded existing custody object; explicit reconciliation required')
|
|
objects[name]=identity(pin.fd);self.commit('created-'+name)
|
|
finally:pin.close()
|
|
fd=secure(path,True)
|
|
try:require(identity(fd)==objects[name],'custody directory replaced: '+name)
|
|
finally:os.close(fd)
|
|
|
|
def bind(self,name,source_fd,target):
|
|
count=mounted(target)
|
|
expected=identity(source_fd)
|
|
if count:
|
|
require(count==1,'stacked/ambiguous custody mounts')
|
|
fd=secure(target,True)
|
|
try:require(identity(fd)==expected,'foreign custody mount')
|
|
finally:os.close(fd)
|
|
else:
|
|
self.directory(name,target)
|
|
# Open and retain the protected target inode too; mount must not
|
|
# resolve mutable source ancestry after the source pin was captured.
|
|
target_fd=secure(target,True)
|
|
try:
|
|
self.commit('bind-'+name+'-intent')
|
|
if LIBC.mount(os.fsencode('/proc/self/fd/'+str(source_fd)),
|
|
os.fsencode('/proc/self/fd/'+str(target_fd)),None,MS_BIND,None):
|
|
raise OSError(ctypes.get_errno(),'bind pinned custody directory')
|
|
finally:os.close(target_fd)
|
|
if LIBC.mount(None,os.fsencode(target),None,MS_PRIVATE,None):
|
|
raise OSError(ctypes.get_errno(),'private custody propagation')
|
|
fd=secure(target,True)
|
|
try:require(identity(fd)==expected,'custody bind postimage differs')
|
|
finally:os.close(fd)
|
|
self.commit('bound-'+name)
|
|
|
|
def recover(self):
|
|
custody.refuse_legacy()
|
|
source=pin_source(self.value['source'],self.value['source_identity']);self.pins.append(source)
|
|
alias=Path(self.value['alias']);os_root=Path(self.value['os_root'])
|
|
self.bind('support-mountpoint',source,alias)
|
|
stage=alias/self.value['stage_name']
|
|
self.directory('stage',stage)
|
|
stage_pin=custody.Pin(stage);self.pins.append(stage_pin.fd)
|
|
stage_fd=os.dup(stage_pin.fd);stage_pin.close()
|
|
self.pins[-1]=stage_fd
|
|
self.bind('os-mountpoint',stage_fd,os_root)
|
|
self.check()
|
|
self.commit('ready')
|
|
return self.receipt()
|
|
|
|
def check(self):
|
|
require(read(self.config)==self.raw,'OS custody appointment changed')
|
|
self.state_pin.check()
|
|
named=secure(self.state/'owner.lock')
|
|
try:require(identity(named)==self.lock_identity==identity(self.lock),'OS custody owner lock replaced')
|
|
finally:os.close(named)
|
|
require(json.loads(read(self.journal))==self.head,'OS custody journal replaced')
|
|
source=pin_source(self.value['source'],self.value['source_identity'])
|
|
try:
|
|
for path,expected in ((self.value['alias'],identity(source)),
|
|
(self.value['os_root'],self.head['objects']['stage'])):
|
|
require(mounted(path)==1,'OS custody mount missing/stacked')
|
|
fd=secure(path,True)
|
|
try:require(identity(fd)==expected,'OS custody mount inode differs')
|
|
finally:os.close(fd)
|
|
fd=secure(Path(self.value['alias'])/self.value['stage_name'],True)
|
|
try:require(identity(fd)==self.head['objects']['stage'],'staging custody replaced')
|
|
finally:os.close(fd)
|
|
finally:os.close(source)
|
|
|
|
def receipt(self):
|
|
return {'configuration_sha256':self.digest,'journal_sha256':hashlib.sha256(read(self.journal)).hexdigest(),
|
|
'sequence':self.head['sequence'],'phase':self.head['phase'],'node':self.value['node'],
|
|
'source_identity':self.value['source_identity'],'stage_identity':self.head['objects']['stage'],
|
|
'os_root':self.value['os_root'],'alias':self.value['alias'],'source_ownership_changed':False}
|
|
|
|
def close(self):
|
|
for fd in self.pins:os.close(fd)
|
|
self.pins=[]
|
|
if self.lock is not None:os.close(self.lock);self.lock=None
|
|
if self.state_pin is not None:self.state_pin.close();self.state_pin=None
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__);p.add_argument('operation',choices=('recover','serve'))
|
|
args=p.parse_args();owner=Coordinator()
|
|
try:
|
|
print(json.dumps(owner.recover(),sort_keys=True),flush=True)
|
|
if args.operation=='serve':
|
|
stopped=False
|
|
def stop(*_):
|
|
nonlocal stopped
|
|
stopped=True
|
|
signal.signal(signal.SIGTERM,stop);signal.signal(signal.SIGINT,stop)
|
|
address=os.environ.get('NOTIFY_SOCKET')
|
|
if address:
|
|
with socket.socket(socket.AF_UNIX,socket.SOCK_DGRAM) as channel:
|
|
channel.connect('\0'+address[1:] if address.startswith('@') else address)
|
|
channel.send(b'READY=1\nSTATUS=Root-owned data custody pinned and verified')
|
|
while not stopped:owner.check();time.sleep(.25)
|
|
finally:owner.close()
|
|
|
|
if __name__=='__main__':main()
|