#!/usr/bin/env python3 """Operator-run on RustFS only. Never logs credentials; does not provision buckets. Requires root access to /opt/rustfs/.env. Refuses to overwrite existing identities, policies, or credential files. Credential delivery file is owner-only for pmp. """ import argparse import datetime import hashlib import hmac import json import os from pathlib import Path import pwd import re import secrets import shlex import urllib.error import urllib.parse import urllib.request import xml.etree.ElementTree as ET class Admin: def __init__(self): values = {} for line in Path('/opt/rustfs/.env').read_text().splitlines(): match = re.fullmatch(r'\s*(?:export\s+)?([A-Za-z_][A-Za-z_0-9]*)\s*=\s*(.*)', line) if match: parts = shlex.split(match.group(2), comments=True) if len(parts) <= 1: values[match.group(1)] = parts[0] if parts else '' self.key = values['RUSTFS_ACCESS_KEY'] self.secret = values['RUSTFS_SECRET_KEY'] self.endpoint = 'http://127.0.0.1:9000' def request(self, method, path, query=None, body=b'', key=None, secret=None, content_type='application/json'): key = self.key if key is None else key secret = self.secret if secret is None else secret query_string = '&'.join(urllib.parse.quote(str(k), safe='-_.~') + '=' + urllib.parse.quote(str(v), safe='-_.~') for k, v in sorted((query or {}).items())) now = datetime.datetime.now(datetime.timezone.utc) date, instant = now.strftime('%Y%m%d'), now.strftime('%Y%m%dT%H%M%SZ') payload_hash = hashlib.sha256(body).hexdigest() headers = {'host': urllib.parse.urlsplit(self.endpoint).netloc, 'x-amz-content-sha256': payload_hash, 'x-amz-date': instant} signed_headers = ';'.join(sorted(headers)) canonical_headers = ''.join(k + ':' + headers[k] + '\n' for k in sorted(headers)) canonical_request = '\n'.join((method, path, query_string, canonical_headers, signed_headers, payload_hash)) scope = date + '/us-east-1/s3/aws4_request' to_sign = '\n'.join(('AWS4-HMAC-SHA256', instant, scope, hashlib.sha256(canonical_request.encode()).hexdigest())) signing_key = ('AWS4' + secret).encode() for item in (date, 'us-east-1', 's3', 'aws4_request'): signing_key = hmac.new(signing_key, item.encode(), hashlib.sha256).digest() signature = hmac.new(signing_key, to_sign.encode(), hashlib.sha256).hexdigest() headers['Authorization'] = ('AWS4-HMAC-SHA256 Credential=' + key + '/' + scope + ', SignedHeaders=' + signed_headers + ', Signature=' + signature) if body: headers['Content-Type'] = content_type url = self.endpoint + path + ('?' + query_string if query_string else '') req = urllib.request.Request(url, data=body if method in ('PUT', 'POST') else None, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=30) as response: return response.status, response.read() except urllib.error.HTTPError as error: return error.code, error.read() def require(status, data, expected, operation): if status not in expected: code = 'unknown' try: code = ET.fromstring(data).findtext('Code') or 'unknown' except Exception: pass raise SystemExit(operation + ' failed: HTTP ' + str(status) + ', code ' + code) def create(policy_path): username, policy_name, bucket = 'btcdata-archive', 'btcdata-archive-runtime-v1', 'btcdata-archive' target = Path('/home/pmp/.config/btcdata-archive') path = target / 's3.env' if path.exists(): raise SystemExit('Credential delivery file already exists; refusing to overwrite') policy = json.loads(policy_path.read_text()) admin = Admin() status, data = admin.request('GET', '/rustfs/admin/v3/list-users') require(status, data, {200}, 'Administrative authentication') users = json.loads(data) if not isinstance(users, dict): raise SystemExit('Unexpected user listing format') if username in users: raise SystemExit('Identity already exists; refusing to overwrite or rotate it') status, data = admin.request('GET', '/rustfs/admin/v3/list-canned-policies') require(status, data, {200}, 'Policy listing') policies = json.loads(data) if not isinstance(policies, dict): raise SystemExit('Unexpected policy listing format') if policy_name in policies: raise SystemExit('Named policy already exists; refusing to overwrite it') status, data = admin.request('HEAD', '/' + bucket) require(status, data, {200, 404}, 'Archive bucket inspection') bucket_exists = status == 200 print('Archive bucket present: ' + str(bucket_exists), flush=True) status, data = admin.request('PUT', '/rustfs/admin/v3/add-canned-policy', {'name': policy_name}, json.dumps(policy, separators=(',', ':')).encode()) require(status, data, {200, 204}, 'Restricted policy creation') secret = secrets.token_urlsafe(40) created = False try: status, data = admin.request('PUT', '/rustfs/admin/v3/add-user', {'accessKey': username}, json.dumps({'secretKey': secret, 'status': 'enabled'}).encode()) require(status, data, {200, 204}, 'Identity creation') created = True status, data = admin.request('PUT', '/rustfs/admin/v3/set-user-or-group-policy', {'policyName': policy_name, 'userOrGroup': username, 'isGroup': 'false'}) require(status, data, {200, 204}, 'Policy attachment') status, data = admin.request('GET', '/rustfs/admin/v3/user-info', {'accessKey': username}) require(status, data, {200}, 'Identity verification') info = json.loads(data) if info.get('policyName') != policy_name: raise SystemExit('Unexpected policy binding; refusing credential delivery') checks = [ ('other bucket access', 'GET', '/xtdb', {'list-type': '2', 'prefix': 'btcvault/', 'max-keys': '1'}), ('administrative access', 'GET', '/rustfs/admin/v3/list-users', {}), ('self-admin access', 'GET', '/rustfs/admin/v3/user-info', {'accessKey': username}), ('object deletion', 'DELETE', '/xtdb/.identity-deny-probe-' + secrets.token_hex(16), {}), ] if bucket_exists: checks.append(('outside-prefix access', 'GET', '/' + bucket, {'list-type': '2', 'prefix': 'outside/', 'max-keys': '1'})) for label, method, route, query in checks: status, data = admin.request(method, route, query, key=username, secret=secret) require(status, data, {403}, 'Deny check: ' + label) print('Verified denied: ' + label, flush=True) if bucket_exists: status, data = admin.request('GET', '/' + bucket, {'list-type': '2', 'prefix': 'btcprod/weeks/', 'max-keys': '1'}, key=username, secret=secret) require(status, data, {200}, 'Allowed-prefix access') print('Verified allowed: archive prefix listing', flush=True) else: print('Positive archive access checks pending bucket provisioning', flush=True) target.mkdir(mode=0o700, parents=True, exist_ok=True) identity = pwd.getpwnam('pmp') os.chmod(target, 0o700) os.chown(target, identity.pw_uid, identity.pw_gid) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, 'w') as output: output.write('AWS_ACCESS_KEY_ID=' + shlex.quote(username) + '\nAWS_SECRET_ACCESS_KEY=' + shlex.quote(secret) + '\n') os.chown(path, identity.pw_uid, identity.pw_gid) print('Identity created; credentials saved owner-only for secure transfer. No secret values displayed.', flush=True) except BaseException: if created: status, data = admin.request('PUT', '/rustfs/admin/v3/set-user-status', {'accessKey': username, 'status': 'disabled'}) print('Identity disabled after incomplete provisioning: ' + str(status in (200, 204)), flush=True) raise if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('policy_file', type=Path) args = parser.parse_args() create(args.policy_file)