net-kingdom/tools/bootstrap-recovery/kingpath-offsite.py

#!/usr/bin/env python3
"""Seal break-glass access material to the backup recipient and publish it to Storage Share.

NK-WP-0055-T13. The founder runs this in their own terminal. Named private files
(admin kubeconfig, static tunnel key, ops bundle) are packed in memory, encrypted
to the registered public backup recipient, uploaded under a new name to the
`Backup` account and read back with a hash check. Plaintext never reaches disk;
the receipt holds names, sizes and the ciphertext digest only.
"""
import argparse
import base64
from datetime import datetime,timezone
import getpass
import hashlib
import importlib.util
import io
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tarfile
import tempfile
import urllib.error
import urllib.request
import warnings

HERE=Path(__file__).resolve().parent
spec=importlib.util.spec_from_file_location('identity_offsite',HERE/'identity-offsite.py')
offsite=importlib.util.module_from_spec(spec);spec.loader.exec_module(offsite)
RECIPIENT=json.loads((HERE/'identity-backup-profile.json').read_text())['public_recipient']
FOLDER='railiance-backups/net-kingdom-kingpath/'
ACCOUNT_BASE=offsite.HOST+'/remote.php/dav/files/Backup/'+FOLDER
ITEM=re.compile(r'[a-z0-9][a-z0-9-]{0,39}\Z')
ITEM_LIMIT=64*1024*1024
MAX_ITEMS=8


class Refused(Exception):pass


def items(specs):
    # Each item is a private, regular, single-link file owned by the caller.
    if not 0<len(specs)<=MAX_ITEMS:raise Refused('items')
    result={}
    for spec in specs:
        name,sep,path=spec.partition('=')
        if not sep or not ITEM.fullmatch(name) or name in result:raise Refused('item_name')
        source=Path(os.path.expanduser(path))
        try:info=source.lstat()
        except OSError:raise Refused('item_missing') from None
        if (not source.is_file() or source.is_symlink() or info.st_uid!=os.getuid() or info.st_nlink!=1
            or info.st_mode&0o077 or not 0<info.st_size<=ITEM_LIMIT):raise Refused('item_unsafe')
        result[name]=source
    return result


def bundle(sources):
    # Normalized in-memory tar: no owner names, times or modes from the workstation.
    raw=io.BytesIO()
    with tarfile.open(fileobj=raw,mode='w',format=tarfile.PAX_FORMAT) as archive:
        for name,source in sorted(sources.items()):
            data=source.read_bytes()
            entry=tarfile.TarInfo(name);entry.size=len(data);entry.mode=0o600;entry.mtime=0
            archive.addfile(entry,io.BytesIO(data))
    return raw.getvalue()


def seal(plaintext,target,recipient=RECIPIENT):
    if not re.fullmatch(r'age1[0-9a-z]{58}',recipient):raise Refused('recipient')
    fd=os.open(target,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
    with os.fdopen(fd,'wb') as out:
        done=subprocess.run(['age','-r',recipient],input=plaintext,stdout=out,stderr=subprocess.DEVNULL,check=False)
        out.flush();os.fsync(out.fileno())
    with target.open('rb') as sealed:
        if done.returncode or sealed.read(22)!=b'age-encryption.org/v1\n':raise Refused('seal')
    return offsite.digest(target),target.stat().st_size


def auth(password):
    if not password or any(c in password for c in '\r\n\x00') or len(password)>256:raise Refused('credential')
    return {'Authorization':'Basic '+base64.b64encode(('Backup:'+password).encode()).decode(),
            'X-Requested-With':'XMLHttpRequest','Accept-Encoding':'identity'}


def ensure_folder(client,h):
    try:
        with client.open(urllib.request.Request(ACCOUNT_BASE,method='MKCOL',headers=h),timeout=30) as response:
            if response.status!=201:raise Refused('folder')
    except urllib.error.HTTPError as error:
        if error.code!=405:raise Refused('folder') from None  # 405: the folder already exists


def check(response,url,size):
    value=response.headers.get('ETag','')
    if (response.status!=200 or response.url!=url or response.headers.get('Content-Length')!=str(size)
        or not re.fullmatch(r'"[^"\x00-\x20]{1,180}"',value)):raise Refused('readback')
    return value


def publish(client,password,sealed,sha,size,stamp):
    name=f'net-kingdom-kingpath-{stamp}-{sha[:12]}.tar.age'
    url=ACCOUNT_BASE+name;h=auth(password)
    ensure_folder(client,h)
    try:
        with client.open(urllib.request.Request(url,method='HEAD',headers=h),timeout=30):raise Refused('exists')
    except urllib.error.HTTPError as error:
        if error.code!=404:raise Refused('exists') from None
    put=dict(h);put.update({'If-None-Match':'*','Content-Type':'application/octet-stream','Content-Length':str(size)})
    # An unknown upload outcome is never retried under this name automatically.
    with sealed.open('rb') as stream:
        with client.open(urllib.request.Request(url,data=stream,method='PUT',headers=put),timeout=600) as response:
            if response.status!=201 or response.url!=url:raise Refused('upload')
    with client.open(urllib.request.Request(url,method='HEAD',headers=h),timeout=30) as response:version=check(response,url,size)
    get=dict(h);get['If-Match']=version;digest=hashlib.sha256();count=0
    with client.open(urllib.request.Request(url,headers=get),timeout=180) as response:
        if check(response,url,size)!=version:raise Refused('readback')
        while block:=response.read(1024*1024):
            count+=len(block)
            if count>size:raise Refused('readback')
            digest.update(block)
    if count!=size or digest.hexdigest()!=sha:raise Refused('readback')
    return name,version


def run(specs,password,receipt,client=None,recipient=RECIPIENT,now=None):
    if receipt.exists() or receipt.is_symlink():raise Refused('receipt_exists')
    sources=items(specs)
    stamp=(now or datetime.now(timezone.utc)).strftime('%Y%m%dT%H%M%SZ')
    with tempfile.TemporaryDirectory(prefix='.kingpath-') as temporary:
        sealed=Path(temporary)/'bundle.tar.age'
        sha,size=seal(bundle(sources),sealed,recipient)
        name,version=publish(client or offsite.opener(),password,sealed,sha,size,stamp)
    record={'schema':'netkingdom.kingpath-offsite.v1','status':'readback_verified',
            'observed_at':datetime.now(timezone.utc).isoformat(),'provider':'Storage Share',
            'folder':'Backup/'+FOLDER,'artifact':name,'etag':version,'recipient':recipient,
            'ciphertext_sha256':sha,'ciphertext_bytes':size,
            'items':{n:s.stat().st_size for n,s in sorted(sources.items())},
            'plaintext_written_to_disk':False,'local_ciphertext_removed':True,
            'secret_values_exported':False,'openbao_contacted':False,'sso_contacted':False}
    fd=os.open(receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
    with os.fdopen(fd,'w') as out:json.dump(record,out,indent=2,sort_keys=True);out.write('\n')
    return record


def main():
    parser=argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument('--item',action='append',default=[],metavar='NAME=PATH',required=True)
    parser.add_argument('--receipt',type=Path,required=True)
    args=parser.parse_args()
    if not sys.stdin.isatty():
        print('refused: run in your own terminal',file=sys.stderr);return 2
    try:
        sources=items(args.item)  # refuse unsafe input before asking for the credential
        with warnings.catch_warnings():
            warnings.simplefilter('error',getpass.GetPassWarning)
            password=getpass.getpass('Storage Share Backup app password: ')
        run(args.item,password,args.receipt.expanduser())
    except (Refused,getpass.GetPassWarning) as error:
        print(f'refused: {error}',file=sys.stderr);return 1
    except (urllib.error.URLError,OSError):
        print('refused: provider or local I/O failure (no retry under the same name)',file=sys.stderr);return 1
    print(f'OK: {len(sources)} items sealed, uploaded and read back; receipt {args.receipt}')
    return 0


if __name__=='__main__':sys.exit(main())