2f26f2d836
- onboard-client.py: client replica scaffolding CLI - gws/: Google Workspace sync (Gmail, Calendar, Drive) - lifestream/: life event stream collector - muscriptor-mcp/: audio → MIDI MCP server - music-mcp/: music library MCP server - data_sharing/: consent-gated data sharing (Python + TS) - sync-mirrors.py: GitHub → Forgejo mirror engine - brain-to-gbrain.py, vault-sync.py, test-all.sh - shared/: TS data-sharing library + index - dirac: provider registry update - .gitignore: exclude Rust build artifacts Co-authored-by: Álvaro de Campos <campos@portugalfuturista.org>
402 lines
16 KiB
Python
402 lines
16 KiB
Python
"""
|
|
Transport backends — OS-agnostic data delivery to Portugal Futurista.
|
|
|
|
Each transport is a strategy with a single method:
|
|
|
|
transmit(payload: dict, consent: ConsentRecord) -> TransmissionResult
|
|
|
|
The transport is selected from config.toml [data_sharing].transport:
|
|
|
|
"http" — POST JSON to an HTTP endpoint (works everywhere)
|
|
"ssh" — scp/tar over SSH to a remote host (no Proxmox needed)
|
|
"local" — write to a local directory (testing / air-gapped)
|
|
"s3" — upload to S3-compatible storage (MinIO, AWS, etc.)
|
|
"proxmox" — legacy pct push/pull (Proxmox VE only)
|
|
|
|
All transports are auto-detected from environment + config — no hard-coded IPs,
|
|
container IDs, or OS-specific commands. The same config works on macOS, Linux,
|
|
WSL, Docker, bare metal, or VMs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import urllib.request
|
|
import urllib.error
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Protocol
|
|
|
|
from .consent import ConsentRecord, redact
|
|
|
|
|
|
@dataclass
|
|
class TransmissionResult:
|
|
"""Result of a transmission attempt."""
|
|
success: bool
|
|
transport: str
|
|
message: str
|
|
bytes_sent: int = 0
|
|
timestamp: str = ""
|
|
detail: str = ""
|
|
|
|
|
|
class Transport(Protocol):
|
|
"""Interface every transport backend must implement."""
|
|
|
|
def transmit(self, payload: dict[str, Any], consent: ConsentRecord) -> TransmissionResult:
|
|
...
|
|
|
|
def name(self) -> str:
|
|
...
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# HTTP Transport — universal, works on any OS with Python stdlib
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class HttpTransport:
|
|
"""
|
|
POST JSON payload to an HTTP endpoint.
|
|
|
|
Config keys (in [data_sharing]):
|
|
endpoint: URL to POST to (required)
|
|
auth_token_env: env var name for bearer token (optional)
|
|
timeout: seconds (default 30)
|
|
|
|
Env overrides:
|
|
DATA_SHARING_ENDPOINT: overrides endpoint
|
|
DATA_SHARING_TOKEN: bearer token (if auth_token_env not set)
|
|
"""
|
|
|
|
def name(self) -> str:
|
|
return "http"
|
|
|
|
def transmit(self, payload: dict[str, Any], consent: ConsentRecord) -> TransmissionResult:
|
|
endpoint = (
|
|
os.environ.get("DATA_SHARING_ENDPOINT")
|
|
or consent.endpoint
|
|
)
|
|
if not endpoint:
|
|
return TransmissionResult(
|
|
False, "http", "No endpoint configured (set [data_sharing].endpoint or DATA_SHARING_ENDPOINT)"
|
|
)
|
|
|
|
timeout = 30
|
|
if consent._raw and "timeout" in (consent._raw or {}):
|
|
timeout = int(consent._raw["timeout"])
|
|
|
|
# Resolve auth token
|
|
token = os.environ.get("DATA_SHARING_TOKEN", "")
|
|
auth_env = ""
|
|
if consent._raw and "auth_token_env" in (consent._raw or {}):
|
|
auth_env = consent._raw["auth_token_env"]
|
|
token = os.environ.get(auth_env, token)
|
|
|
|
# Serialize + redact
|
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
if consent.redact_secrets:
|
|
body_str = body.decode("utf-8")
|
|
body_str = redact(body_str)
|
|
body = body_str.encode("utf-8")
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Aurelio-Source": consent._raw.get("client_id", "replica") if consent._raw else "replica",
|
|
"X-Aurelio-Transport": "http",
|
|
}
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
|
|
try:
|
|
req = urllib.request.Request(endpoint, data=body, headers=headers, method="POST")
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return TransmissionResult(
|
|
success=200 <= resp.status < 300,
|
|
transport="http",
|
|
message=f"HTTP {resp.status}",
|
|
bytes_sent=len(body),
|
|
timestamp=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
except urllib.error.HTTPError as e:
|
|
return TransmissionResult(False, "http", f"HTTP {e.code}: {e.reason}", len(body))
|
|
except Exception as e:
|
|
return TransmissionResult(False, "http", f"Request failed: {e}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# SSH Transport — works on any OS with ssh+scp in PATH
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class SshTransport:
|
|
"""
|
|
Upload a tarball of the payload via SSH + scp.
|
|
|
|
Config keys:
|
|
ssh_host: user@host (required)
|
|
ssh_path: remote directory (default /opt/aurelio-ingest)
|
|
ssh_port: port (default 22)
|
|
|
|
Env overrides:
|
|
DATA_SHARING_SSH_HOST
|
|
DATA_SHARING_SSH_PATH
|
|
DATA_SHARING_SSH_PORT
|
|
"""
|
|
|
|
def name(self) -> str:
|
|
return "ssh"
|
|
|
|
def transmit(self, payload: dict[str, Any], consent: ConsentRecord) -> TransmissionResult:
|
|
host = os.environ.get("DATA_SHARING_SSH_HOST", "")
|
|
remote_path = os.environ.get("DATA_SHARING_SSH_PATH", "/opt/aurelio-ingest")
|
|
port = os.environ.get("DATA_SHARING_SSH_PORT", "22")
|
|
|
|
raw = consent._raw or {}
|
|
host = raw.get("ssh_host", host)
|
|
remote_path = raw.get("ssh_path", remote_path)
|
|
port = str(raw.get("ssh_port", port))
|
|
|
|
if not host:
|
|
return TransmissionResult(False, "ssh", "No ssh_host configured")
|
|
|
|
if not shutil.which("ssh") or not shutil.which("scp"):
|
|
return TransmissionResult(False, "ssh", "ssh/scp not in PATH")
|
|
|
|
try:
|
|
with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as tmp:
|
|
tmp_path = tmp.name
|
|
|
|
with tarfile.open(tmp_path, "w:gz") as tar:
|
|
data = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
|
|
if consent.redact_secrets:
|
|
data = redact(data.decode("utf-8")).encode("utf-8")
|
|
import io
|
|
info = tarfile.TarInfo(name="payload.json")
|
|
info.size = len(data)
|
|
tar.addfile(info, io.BytesIO(data))
|
|
|
|
# scp
|
|
scp_cmd = ["scp", "-P", port, tmp_path, f"{host}:{remote_path}/"]
|
|
r = subprocess.run(scp_cmd, capture_output=True, timeout=60)
|
|
os.unlink(tmp_path)
|
|
|
|
if r.returncode != 0:
|
|
return TransmissionResult(
|
|
False, "ssh", f"scp failed: {r.stderr.decode()[:200]}"
|
|
)
|
|
|
|
return TransmissionResult(
|
|
True, "ssh", f"Uploaded to {host}:{remote_path}",
|
|
bytes_sent=os.path.getsize(tmp_path) if os.path.exists(tmp_path) else 0,
|
|
timestamp=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
except Exception as e:
|
|
return TransmissionResult(False, "ssh", f"SSH transmit failed: {e}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Local Transport — testing / air-gapped environments
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class LocalTransport:
|
|
"""
|
|
Write payload as JSON to a local directory.
|
|
|
|
Config keys:
|
|
local_path: directory to write to (required)
|
|
"""
|
|
|
|
def name(self) -> str:
|
|
return "local"
|
|
|
|
def transmit(self, payload: dict[str, Any], consent: ConsentRecord) -> TransmissionResult:
|
|
raw = consent._raw or {}
|
|
dest = raw.get("local_path", os.environ.get("DATA_SHARING_LOCAL_PATH", ""))
|
|
|
|
if not dest:
|
|
return TransmissionResult(False, "local", "No local_path configured")
|
|
|
|
dest_path = Path(dest)
|
|
dest_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
out_file = dest_path / f"ingest-{ts}.json"
|
|
|
|
data = json.dumps(payload, ensure_ascii=False, indent=2)
|
|
if consent.redact_secrets:
|
|
data = redact(data)
|
|
|
|
out_file.write_text(data)
|
|
size = out_file.stat().st_size
|
|
|
|
return TransmissionResult(
|
|
True, "local", f"Written to {out_file}",
|
|
bytes_sent=size,
|
|
timestamp=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
|
|
|
|
# ╁══════════════════════════════════════════════════════════════════════
|
|
# S3 Transport — MinIO, AWS S3, Cloudflare R2, etc.
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class S3Transport:
|
|
"""
|
|
Upload payload to S3-compatible storage using the REST API (no SDK dep).
|
|
|
|
Config keys:
|
|
s3_endpoint: e.g. https://192.168.0.40:9000 or https://s3.amazonaws.com
|
|
s3_bucket: bucket name
|
|
s3_access_key_env: env var for access key
|
|
s3_secret_key_env: env var for secret key
|
|
"""
|
|
|
|
def name(self) -> str:
|
|
return "s3"
|
|
|
|
def transmit(self, payload: dict[str, Any], consent: ConsentRecord) -> TransmissionResult:
|
|
import hashlib
|
|
import hmac
|
|
import base64
|
|
|
|
raw = consent._raw or {}
|
|
endpoint = raw.get("s3_endpoint", os.environ.get("DATA_SHARING_S3_ENDPOINT", ""))
|
|
bucket = raw.get("s3_bucket", os.environ.get("DATA_SHARING_S3_BUCKET", ""))
|
|
|
|
if not endpoint or not bucket:
|
|
return TransmissionResult(False, "s3", "Missing s3_endpoint or s3_bucket")
|
|
|
|
access_key = os.environ.get(raw.get("s3_access_key_env", "S3_ACCESS_KEY"), "")
|
|
secret_key = os.environ.get(raw.get("s3_secret_key_env", "S3_SECRET_KEY"), "")
|
|
|
|
if not access_key or not secret_key:
|
|
return TransmissionResult(False, "s3", "Missing S3 credentials")
|
|
|
|
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
if consent.redact_secrets:
|
|
data = redact(data.decode("utf-8")).encode("utf-8")
|
|
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
object_key = f"ingest/ingest-{ts}.json"
|
|
|
|
# Simple PUT — no SigV4 (works with MinIO public-write buckets or
|
|
# pre-signed URLs; for AWS S3 use the http transport with a lambda)
|
|
url = f"{endpoint.rstrip('/')}/{bucket}/{object_key}"
|
|
req = urllib.request.Request(url, data=data, method="PUT")
|
|
req.add_header("Content-Type", "application/json")
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return TransmissionResult(
|
|
200 <= resp.status < 300, "s3", f"S3 PUT {resp.status}",
|
|
bytes_sent=len(data),
|
|
timestamp=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
except urllib.error.HTTPError as e:
|
|
return TransmissionResult(False, "s3", f"S3 PUT {e.code}: {e.reason}", len(data))
|
|
except Exception as e:
|
|
return TransmissionResult(False, "s3", f"S3 upload failed: {e}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Proxmox Transport — legacy pct push/pull (Proxmox VE only)
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class ProxmoxTransport:
|
|
"""
|
|
Legacy transport using pct push/pull via a Proxmox host.
|
|
Only works when running ON or WITH ACCESS TO a Proxmox VE node.
|
|
|
|
Config keys:
|
|
proxmox_host: IP/hostname of the PVE node
|
|
proxmox_ct: container ID
|
|
proxmox_path: target path inside the CT (default /opt/aurelio/brain)
|
|
"""
|
|
|
|
def name(self) -> str:
|
|
return "proxmox"
|
|
|
|
def transmit(self, payload: dict[str, Any], consent: ConsentRecord) -> TransmissionResult:
|
|
raw = consent._raw or {}
|
|
host = raw.get("proxmox_host", os.environ.get("PROXMOX_HOST", ""))
|
|
ct_id = raw.get("proxmox_ct", 0)
|
|
ct_path = raw.get("proxmox_path", "/opt/aurelio/brain")
|
|
|
|
if not host or not ct_id:
|
|
return TransmissionResult(False, "proxmox", "Missing proxmox_host or proxmox_ct")
|
|
|
|
if not shutil.which("ssh"):
|
|
return TransmissionResult(False, "proxmox", "ssh not in PATH")
|
|
|
|
try:
|
|
with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as tmp:
|
|
tmp_path = tmp.name
|
|
|
|
with tarfile.open(tmp_path, "w:gz") as tar:
|
|
data = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
|
|
if consent.redact_secrets:
|
|
data = redact(data.decode("utf-8")).encode("utf-8")
|
|
import io
|
|
info = tarfile.TarInfo(name="payload.json")
|
|
info.size = len(data)
|
|
tar.addfile(info, io.BytesIO(data))
|
|
|
|
# scp to proxmox host
|
|
r = subprocess.run(
|
|
["scp", tmp_path, f"root@{host}:/tmp/"],
|
|
capture_output=True, timeout=60
|
|
)
|
|
if r.returncode != 0:
|
|
os.unlink(tmp_path)
|
|
return TransmissionResult(False, "proxmox", f"scp failed: {r.stderr.decode()[:200]}")
|
|
|
|
# pct push into CT
|
|
r = subprocess.run(
|
|
["ssh", f"root@{host}", "pct", "push", str(ct_id),
|
|
f"/tmp/{Path(tmp_path).name}", f"/tmp/{Path(tmp_path).name}"],
|
|
capture_output=True, timeout=60
|
|
)
|
|
os.unlink(tmp_path)
|
|
if r.returncode != 0:
|
|
return TransmissionResult(False, "proxmox", f"pct push failed: {r.stderr.decode()[:200]}")
|
|
|
|
size = os.path.getsize(tmp_path) if os.path.exists(tmp_path) else 0
|
|
return TransmissionResult(
|
|
True, "proxmox", f"Pushed to CT {ct_id} on {host}",
|
|
bytes_sent=size,
|
|
timestamp=datetime.now(timezone.utc).isoformat(),
|
|
)
|
|
except Exception as e:
|
|
return TransmissionResult(False, "proxmox", f"Proxmox transmit failed: {e}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# Registry
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
TRANSPORTS: dict[str, Transport] = {
|
|
"http": HttpTransport(),
|
|
"ssh": SshTransport(),
|
|
"local": LocalTransport(),
|
|
"s3": S3Transport(),
|
|
"proxmox": ProxmoxTransport(),
|
|
}
|
|
|
|
|
|
def get_transport(name: str) -> Transport:
|
|
"""Get a transport backend by name. Raises ValueError if unknown."""
|
|
if name not in TRANSPORTS:
|
|
available = ", ".join(TRANSPORTS)
|
|
raise ValueError(f"Unknown transport '{name}'. Available: {available}")
|
|
return TRANSPORTS[name]
|
|
|
|
|
|
def available_transports() -> list[str]:
|
|
"""Return list of registered transport names."""
|
|
return list(TRANSPORTS.keys())
|