tegwick
c262470089
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
109 lines
5.5 KiB
Python
109 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent read-only inventory in the existing Platform Backup-account procedure.
|
|
|
|
Runs only inside the attended platform-admin envelope. Reuses Platform's owner
|
|
custody accessor and fixed provider/account contract; no age key is read and no
|
|
backup is downloaded, uploaded, deleted, rotated or granted.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
import xml.etree.ElementTree as ET
|
|
|
|
ROOT=Path(__file__).resolve().parents[2]
|
|
OWNER=ROOT.parent/'railiance-platform/scripts'
|
|
sys.path.insert(0,str(OWNER))
|
|
from state_hub_preflight_lane import bao, data
|
|
from migrate_nextcloud_backup_account import HOST, OPERATOR_PATH
|
|
BASE='/remote.php/dav/files/Backup/railiance-backups/'
|
|
LIMIT=4*1024*1024
|
|
NAME=re.compile(r'[A-Za-z0-9._-]{1,160}\Z')
|
|
|
|
|
|
class Refused(Exception):pass
|
|
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self,*args,**kwargs):raise Refused()
|
|
|
|
|
|
def parse_listing(raw, relative):
|
|
if len(raw)>LIMIT:raise Refused()
|
|
rows=[]
|
|
for item in ET.fromstring(raw).findall('{DAV:}response'):
|
|
href=urllib.parse.urlsplit(item.findtext('{DAV:}href') or '')
|
|
if href.netloc and href.netloc!=urllib.parse.urlsplit(HOST).netloc:raise Refused()
|
|
path=urllib.parse.unquote(href.path)
|
|
if not path.startswith(BASE+relative):raise Refused()
|
|
name=path[len(BASE+relative):].rstrip('/')
|
|
if not name:continue
|
|
if not NAME.fullmatch(name) or name in {'.','..'}:raise Refused()
|
|
for row in item.findall('{DAV:}propstat'):
|
|
if ' 200 ' not in (row.findtext('{DAV:}status') or ''):continue
|
|
prop=row.find('{DAV:}prop')
|
|
if prop is None:continue
|
|
folder=prop.find('{DAV:}resourcetype/{DAV:}collection') is not None
|
|
size=prop.findtext('{DAV:}getcontentlength')
|
|
if size is not None and not size.isdigit():raise Refused()
|
|
# Export only known-safe operational names/size, not arbitrary DAV
|
|
# properties, URLs, account metadata or raw provider responses.
|
|
if folder or name.endswith(('.age','.json','.yaml')):
|
|
rows.append({'name':name,'directory':folder,'bytes':int(size) if size else None})
|
|
if len(rows)>10000 or len({r['name'] for r in rows})!=len(rows):raise Refused()
|
|
return sorted(rows,key=lambda x:x['name'])
|
|
|
|
|
|
def listing(relative,credentials):
|
|
if relative and (not relative.endswith('/') or not NAME.fullmatch(relative[:-1])):raise Refused()
|
|
auth=base64.b64encode((credentials['BACKUP_USERNAME']+':'+credentials['BACKUP_PASSWORD']).encode()).decode()
|
|
req=urllib.request.Request(HOST+BASE+relative,
|
|
data=b'<d:propfind xmlns:d="DAV:"><d:prop><d:resourcetype/><d:getcontentlength/></d:prop></d:propfind>',
|
|
method='PROPFIND',headers={'Authorization':'Basic '+auth,'Depth':'1','Content-Type':'application/xml'})
|
|
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect())
|
|
with opener.open(req,timeout=30) as response:
|
|
if response.status!=207 or response.url!=HOST+BASE+relative:raise Refused()
|
|
raw=response.read(LIMIT+1)
|
|
return parse_listing(raw,relative)
|
|
|
|
|
|
def inventory():
|
|
# Lookup payload contains a token: retain only safe caller/role facts.
|
|
identity=data(bao(['token','lookup','-format=json']))['data']
|
|
if 'platform-admin' not in identity.get('policies',[]) or 'root' in identity.get('policies',[]):raise Refused()
|
|
if not str(identity.get('path','')).startswith('auth/netkingdom/'):raise Refused()
|
|
credentials=data(bao(['read','-format=json',OPERATOR_PATH]))['data']['data']
|
|
if credentials.get('BACKUP_USERNAME')!='Backup' or not isinstance(credentials.get('BACKUP_PASSWORD'),str) or not credentials['BACKUP_PASSWORD']:raise Refused()
|
|
root=listing('',credentials)
|
|
folders={}
|
|
for row in root:
|
|
if row['directory'] and re.match(r'^(?:r01-|core-|identity|sso|bootstrap|net-kingdom|openbao)',row['name']):
|
|
if len(folders)>=32:raise Refused()
|
|
folders[row['name']]=listing(row['name']+'/',credentials)
|
|
return {'root':root,'folders':folders,'caller_auth_path':identity['path'],
|
|
'caller_entity_id':identity.get('entity_id'),'scope':'Backup/railiance-backups root and bounded identity/database subfolders; other accounts/folders unassessed'}
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--receipt',required=True)
|
|
a=p.parse_args();fd=os.open(a.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
|
|
r={'schema':'netkingdom.storage-share-inventory.v1','status':'failed','started_at':datetime.now(timezone.utc).isoformat(),
|
|
'read_only':True,'age_keys_read':False,'backup_contents_downloaded':False,'secret_values_exported':False,
|
|
'source_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
|
|
'owner_sources':[{'path':str(OWNER/name),'sha256':hashlib.sha256((OWNER/name).read_bytes()).hexdigest()} for name in ['state_hub_preflight_lane.py','migrate_nextcloud_backup_account.py']]}
|
|
try:r.update(inventory());r['status']='metadata_inventory_complete'
|
|
except Exception:r['error']='inventory_failed_diagnostics_suppressed'
|
|
finally:
|
|
r['finished_at']=datetime.now(timezone.utc).isoformat()
|
|
with os.fdopen(fd,'w') as out:json.dump(r,out,indent=2);out.write('\n')
|
|
return int(r['status']=='failed')
|
|
|
|
|
|
if __name__=='__main__':raise SystemExit(main())
|