roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Wiki Activity

feat: connectivity — registry-driven peer table, sshuttle/proxy egress, egress UI

Browse Source 
The peer table was empty because it was not consulting the peer registry;
now peers are driven by PeerRegistry so the Connectivity page reflects actual
connected cells.

Exit-key handling is unified: all code paths now use the same key derivation
so a store-service exit bridge and a manual WireGuard peer both produce
consistent routing state.

Two new egress exit types are added (sshuttle via SSH tunnel and proxy via
redsocks SOCKS5), wiring through connectivity_manager, egress_manager, and
app.py routes. This lets a cell route its traffic through an SSH host or a
SOCKS5 proxy as an alternative to WireGuard exit nodes.

ServiceStoreManager and ServiceBus updated so the egress lifecycle (install /
uninstall) is cleanly signalled between components.

Connectivity.jsx gains the Service Egress section, letting operators assign
and reassign egress methods from the UI without touching config files.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Dmitrii Iurco
2026-06-10 08:36:15 -04:00
parent cc7a223fdf
commit 6232ef23a9
8 changed files with 1096 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/connectivity_manager.py
+411 -16
View File
@@ -3,19 +3,21 @@
Connectivity Manager for Personal Internet Cell — Phase 5 Extended Connectivity.
Provides per-peer egress routing through alternate exits (WireGuard external,
OpenVPN, Tor) via Linux policy routing (fwmark + ip rule + dedicated routing
tables) and dedicated iptables chains.
OpenVPN, Tor, sshuttle SSH tunnel, upstream proxy) via Linux policy routing
(fwmark + ip rule + dedicated routing tables) and dedicated iptables chains.
Architecture
------------
- A peer's `exit_via` field selects the egress path: "default", "wireguard_ext",
"openvpn", or "tor".
"openvpn", "tor", "sshuttle", or "proxy".
- Each non-default exit type is assigned a unique fwmark and a dedicated routing
table:
wireguard_ext mark 0x10 table 110 iface wg_ext0
openvpn mark 0x20 table 120 iface tun0
tor mark 0x30 table 130 (transparent proxy → 9040)
sshuttle mark 0x40 table 140 (transparent proxy → 12300)
proxy mark 0x50 table 150 (transparent proxy → 12345, redsocks)
- All rules live in dedicated PIC_CONNECTIVITY chains in the `mangle` and `nat`
tables so they can be flushed/rebuilt without touching firewall_manager rules.
@@ -36,6 +38,7 @@ Both are validated to strip / reject hook directives that could execute
arbitrary commands on the host.
"""
import ipaddress
import logging
import os
import re
@@ -61,24 +64,100 @@ _OVPN_FORBIDDEN_DIRECTIVES = (
_NAME_RE = re.compile(r'^[a-z0-9_-]{1,32}$')
# sshuttle / proxy configuration validation
_HOST_RE = re.compile(r'^[A-Za-z0-9]([A-Za-z0-9.-]{0,252}[A-Za-z0-9])?$')
_SSH_USER_RE = re.compile(r'^[a-z_][a-z0-9_-]{0,31}$')
_PROXY_USER_RE = re.compile(r'^[A-Za-z0-9._-]{1,64}$')
# Printable ASCII excluding double quote and backslash — safe inside the
# quoted strings of a redsocks.conf without any escaping ambiguity.
_PROXY_PASSWORD_RE = re.compile(r'^[\x20\x21\x23-\x5B\x5D-\x7E]{1,128}$')
_B64_RE = re.compile(r'^[A-Za-z0-9+/]+={0,2}$')
_KNOWN_HOSTS_HOSTS_RE = re.compile(r'^[A-Za-z0-9\[\]:.,*?_-]{1,512}$')
_SSH_KEYTYPES = (
'ssh-ed25519', 'ssh-rsa',
'ecdsa-sha2-nistp256', 'ecdsa-sha2-nistp384', 'ecdsa-sha2-nistp521',
)
def _contains_strict_hostkey_disable(value: str) -> bool:
"""Detect attempts to disable SSH host-key pinning (any spacing/case)."""
normalized = re.sub(r'\s+', '', value.lower())
return ('stricthostkeychecking=no' in normalized
or 'stricthostkeycheckingno' in normalized)
def _validate_host(host) -> Optional[str]:
"""Return a validated hostname/IP, or None when invalid."""
if not isinstance(host, str):
return None
host = host.strip()
if not host or '..' in host or not _HOST_RE.match(host):
return None
return host
def _validate_port(port) -> Optional[int]:
"""Return a validated TCP port (1-65535), or None when invalid."""
try:
port = int(port)
except (TypeError, ValueError):
return None
if not 1 <= port <= 65535:
return None
return port
class ConnectivityManager(BaseServiceManager):
"""Manages alternate egress paths (extended connectivity) for peers."""
EXIT_TYPES = ("default", "wireguard_ext", "openvpn", "tor")
MARKS = {"wireguard_ext": 0x10, "openvpn": 0x20, "tor": 0x30}
TABLES = {"wireguard_ext": 110, "openvpn": 120, "tor": 130}
EXIT_TYPES = ("default", "wireguard_ext", "openvpn", "tor",
"sshuttle", "proxy")
MARKS = {"wireguard_ext": 0x10, "openvpn": 0x20, "tor": 0x30,
"sshuttle": 0x40, "proxy": 0x50}
TABLES = {"wireguard_ext": 110, "openvpn": 120, "tor": 130,
"sshuttle": 140, "proxy": 150}
IFACES = {"wireguard_ext": "wg_ext0", "openvpn": "tun0"}
TOR_TRANS_PORT = 9040
TOR_DNS_PORT = 5353
SSHUTTLE_PORT = 12300
REDSOCKS_PORT = 12345
# Exits that work as pure iptables REDIRECTs to a local transparent-proxy
# port (no exit interface, no kill-switch interface).
REDIRECT_PORTS = {"tor": TOR_TRANS_PORT, "sshuttle": SSHUTTLE_PORT,
"proxy": REDSOCKS_PORT}
# Store-service ids / container names backing each exit type — used to
# report an exit as configured when it was installed via the Service Store
# rather than through a legacy config upload.
STORE_SERVICE_IDS = {
"wireguard_ext": "wireguard-ext",
"openvpn": "openvpn-client",
"tor": "tor",
"sshuttle": "sshuttle",
"proxy": "proxy",
}
EXIT_CONTAINERS = {
"wireguard_ext": "cell-wg-ext",
"openvpn": "cell-openvpn",
"tor": "cell-tor",
"sshuttle": "cell-sshuttle",
"proxy": "cell-redsocks",
}
# RFC1918 ranges excluded from the sshuttle tunnel by default so cell-local
# and LAN traffic is never tunneled.
RFC1918_SUBNETS = ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")
CONNECTIVITY_CHAIN = 'PIC_CONNECTIVITY'
def __init__(self, config_manager=None, peer_registry=None,
vault_manager=None,
data_dir: str = '/app/data', config_dir: str = '/app/config'):
super().__init__('connectivity', data_dir, config_dir)
self.config_manager = config_manager
self.peer_registry = peer_registry
self.vault_manager = vault_manager
# Connectivity configs live under the per-service data dir so that
# ${PIC_DATA_DIR}/services/<id>/config bind mounts in store compose
@@ -87,8 +166,11 @@ class ConnectivityManager(BaseServiceManager):
services_dir = os.path.join(data_dir, 'services')
self.wireguard_ext_dir = os.path.join(services_dir, 'wireguard-ext', 'config')
self.openvpn_dir = os.path.join(services_dir, 'openvpn-client', 'config')
self.sshuttle_dir = os.path.join(services_dir, 'sshuttle', 'config')
self.proxy_dir = os.path.join(services_dir, 'proxy', 'config')
for d in (self.wireguard_ext_dir, self.openvpn_dir):
for d in (self.wireguard_ext_dir, self.openvpn_dir,
self.sshuttle_dir, self.proxy_dir):
self.safe_makedirs(d)
# One-shot migration from the legacy config_dir/connectivity/ location.
@@ -293,6 +375,263 @@ class ConnectivityManager(BaseServiceManager):
logger.info(f"connectivity: stored {name}.ovpn ({len(cleaned)} bytes)")
return {'ok': True}
def configure_sshuttle(self, cfg: Dict[str, Any]) -> Dict[str, Any]:
"""Validate and store an sshuttle (SSH tunnel) exit configuration.
Requires a pinned host key (a single known_hosts line); rejects any
attempt to disable strict host-key checking. Secrets (private key /
password) are written 0o600 under data/services/sshuttle/config/ and
mirrored into the vault — they are never placed in cell_config.json.
"""
if not isinstance(cfg, dict):
return {'ok': False, 'error': 'config must be a JSON object'}
for value in cfg.values():
if isinstance(value, str) and _contains_strict_hostkey_disable(value):
return {
'ok': False,
'error': 'StrictHostKeyChecking=no is not allowed; '
'a pinned host key (known_hosts line) is required',
}
host = _validate_host(cfg.get('host'))
if not host:
return {'ok': False, 'error': 'invalid host: must be a hostname or IP'}
port = _validate_port(cfg.get('port', 22))
if port is None:
return {'ok': False, 'error': 'invalid port: must be 1-65535'}
user = cfg.get('user')
if not isinstance(user, str) or not _SSH_USER_RE.match(user):
return {
'ok': False,
'error': 'invalid user: must match ^[a-z_][a-z0-9_-]{0,31}$',
}
auth = cfg.get('auth', 'key')
if auth not in ('key', 'password'):
return {'ok': False, 'error': "invalid auth: must be 'key' or 'password'"}
known_hosts = cfg.get('known_hosts')
err = self._validate_known_hosts_line(known_hosts)
if err:
return {'ok': False, 'error': err}
known_hosts = known_hosts.strip()
private_key = ''
password = ''
if auth == 'key':
private_key = cfg.get('private_key', '')
if not isinstance(private_key, str) or 'PRIVATE KEY' not in private_key:
return {
'ok': False,
'error': 'private_key is required for key auth and must be '
'a PEM/OpenSSH private key',
}
else:
password = cfg.get('password', '')
if not isinstance(password, str) or not password or '\n' in password:
return {'ok': False, 'error': 'password is required for password auth'}
exclude_subnets = cfg.get('exclude_subnets')
if exclude_subnets is None:
exclude_subnets = self._default_exclude_subnets()
if not isinstance(exclude_subnets, list):
return {'ok': False, 'error': 'exclude_subnets must be a list of CIDRs'}
validated_excludes = []
for net in exclude_subnets:
try:
validated_excludes.append(str(ipaddress.ip_network(str(net), strict=False)))
except ValueError:
return {'ok': False, 'error': f'invalid exclude subnet: {net!r}'}
conf_lines = [
f'HOST={host}',
f'PORT={port}',
f'USER={user}',
f'AUTH={auth}',
f'LISTEN_PORT={self.SSHUTTLE_PORT}',
f'EXCLUDE={",".join(validated_excludes)}',
]
try:
self._write_secure(
os.path.join(self.sshuttle_dir, 'known_hosts'),
known_hosts + '\n',
)
if auth == 'key':
key_text = private_key.rstrip('\n') + '\n'
self._write_secure(os.path.join(self.sshuttle_dir, 'id_pic'), key_text)
else:
self._write_secure(
os.path.join(self.sshuttle_dir, 'password'), password + '\n')
self._write_secure(
os.path.join(self.sshuttle_dir, 'sshuttle.conf'),
'\n'.join(conf_lines) + '\n',
)
except Exception as e:
logger.error(f"configure_sshuttle: write failed: {e}")
return {'ok': False, 'error': 'failed to write sshuttle configuration'}
if self.vault_manager is not None:
try:
if auth == 'key':
self.vault_manager.store_secret('connectivity_sshuttle_key',
private_key)
else:
self.vault_manager.store_secret('connectivity_sshuttle_password',
password)
except Exception as e:
logger.warning(f"configure_sshuttle: vault store failed: {e}")
self._persist_exit_config('sshuttle', {
'host': host,
'port': port,
'user': user,
'auth': auth,
'exclude_subnets': validated_excludes,
})
logger.info(f"connectivity: configured sshuttle exit ({user}@{host}:{port})")
return {'ok': True}
def configure_proxy(self, cfg: Dict[str, Any]) -> Dict[str, Any]:
"""Validate and store an upstream proxy (redsocks) exit configuration.
Generates redsocks.conf from strictly validated fields only — no value
that could break out of the quoted config strings is accepted. The
password lives in the 0o600 conf file, never in compose env.
"""
if not isinstance(cfg, dict):
return {'ok': False, 'error': 'config must be a JSON object'}
scheme = cfg.get('scheme')
if scheme not in ('http', 'socks5'):
return {'ok': False, 'error': "invalid scheme: must be 'http' or 'socks5'"}
host = _validate_host(cfg.get('host'))
if not host:
return {'ok': False, 'error': 'invalid host: must be a hostname or IP'}
port = _validate_port(cfg.get('port'))
if port is None:
return {'ok': False, 'error': 'invalid port: must be 1-65535'}
user = cfg.get('user') or ''
password = cfg.get('password') or ''
if user and not (isinstance(user, str) and _PROXY_USER_RE.match(user)):
return {
'ok': False,
'error': 'invalid user: must match ^[A-Za-z0-9._-]{1,64}$',
}
if password and not (isinstance(password, str)
and _PROXY_PASSWORD_RE.match(password)):
return {
'ok': False,
'error': 'invalid password: 1-128 printable ASCII characters '
'excluding double quotes and backslashes',
}
if password and not user:
return {'ok': False, 'error': 'password requires a user'}
conf = self._render_redsocks_conf(scheme, host, port, user, password)
try:
self._write_secure(os.path.join(self.proxy_dir, 'redsocks.conf'), conf)
except Exception as e:
logger.error(f"configure_proxy: write failed: {e}")
return {'ok': False, 'error': 'failed to write redsocks configuration'}
self._persist_exit_config('proxy', {
'scheme': scheme,
'host': host,
'port': port,
'user': user,
})
logger.info(f"connectivity: configured proxy exit ({scheme}://{host}:{port})")
return {'ok': True}
def _render_redsocks_conf(self, scheme: str, host: str, port: int,
user: str, password: str) -> str:
"""Build a redsocks.conf from already-validated fields."""
redsocks_type = 'socks5' if scheme == 'socks5' else 'http-connect'
lines = [
'base {',
' log_debug = off;',
' log_info = on;',
' log = stderr;',
' daemon = off;',
' redirector = iptables;',
'}',
'',
'redsocks {',
' local_ip = 0.0.0.0;',
f' local_port = {self.REDSOCKS_PORT};',
f' ip = {host};',
f' port = {port};',
f' type = {redsocks_type};',
]
if user:
lines.append(f' login = "{user}";')
if password:
lines.append(f' password = "{password}";')
lines.append('}')
return '\n'.join(lines) + '\n'
@staticmethod
def _validate_known_hosts_line(line) -> Optional[str]:
"""Validate a single known_hosts line; return an error string or None.
Expected format: host[,ip] keytype base64key [comment]
"""
if not isinstance(line, str) or not line.strip():
return ('known_hosts is required: a pinned host key line '
'(host[,ip] keytype base64key)')
line = line.strip()
if '\n' in line or '\r' in line:
return 'known_hosts must be a single line'
parts = line.split()
if len(parts) < 3:
return ('invalid known_hosts line: expected '
'host[,ip] keytype base64key')
hosts, keytype, key = parts[0], parts[1], parts[2]
if not _KNOWN_HOSTS_HOSTS_RE.match(hosts):
return f'invalid known_hosts host field: {hosts!r}'
if keytype not in _SSH_KEYTYPES:
return (f'invalid known_hosts key type {keytype!r}; '
f'must be one of {_SSH_KEYTYPES}')
if not _B64_RE.match(key):
return 'invalid known_hosts key: not valid base64'
return None
def _default_exclude_subnets(self) -> List[str]:
"""Cell subnet + RFC1918 ranges — internal traffic is never tunneled."""
excludes = list(self.RFC1918_SUBNETS)
try:
if self.config_manager is not None:
identity = self.config_manager.get_identity()
if isinstance(identity, dict):
ip_range = identity.get('ip_range', '')
if isinstance(ip_range, str) and '/' in ip_range \
and ip_range not in excludes:
excludes.insert(0, ip_range)
except Exception as e:
logger.debug(f"_default_exclude_subnets: {e}")
return excludes
def _persist_exit_config(self, exit_type: str, fields: Dict[str, Any]) -> None:
"""Persist non-secret exit fields under connectivity.exits in config."""
if self.config_manager is None:
return
try:
cfg = self.config_manager.get_connectivity_config()
exits = cfg.get('exits') if isinstance(cfg, dict) else None
exits = dict(exits) if isinstance(exits, dict) else {}
exits[exit_type] = fields
self.config_manager.set_connectivity_field('exits', exits)
except Exception as e:
logger.warning(f"_persist_exit_config({exit_type}): {e}")
# ── Routing application ───────────────────────────────────────────────
def apply_routes(self) -> Dict[str, Any]:
@@ -313,7 +652,7 @@ class ConnectivityManager(BaseServiceManager):
logger.warning(f"apply_routes: flush {table}/{chain} failed: {e}")
# Idempotent ip rule registration for each non-default exit
for exit_type in ('wireguard_ext', 'openvpn', 'tor'):
for exit_type in self.MARKS:
mark = self.MARKS[exit_type]
table = self.TABLES[exit_type]
try:
@@ -347,14 +686,15 @@ class ConnectivityManager(BaseServiceManager):
f"apply_routes: mark rule for {src_ip}/{exit_via}: {e}"
)
# Tor: redirect TCP to local transparent proxy
if exit_via == 'tor':
# Tor / sshuttle / proxy: redirect TCP to the local
# transparent-proxy port for that exit.
if exit_via in self.REDIRECT_PORTS:
try:
self._add_tor_redirect(src_ip)
self._add_redirect(src_ip, self.REDIRECT_PORTS[exit_via])
rules_applied += 1
except Exception as e:
logger.warning(
f"apply_routes: tor redirect for {src_ip}: {e}"
f"apply_routes: {exit_via} redirect for {src_ip}: {e}"
)
# Kill-switch: drop marked packets that would otherwise leak via the
@@ -435,14 +775,18 @@ class ConnectivityManager(BaseServiceManager):
'-j', 'MARK', '--set-mark', hex(mark),
])
def _add_tor_redirect(self, src_ip: str) -> None:
"""Redirect peer's TCP traffic to local Tor TransPort."""
def _add_redirect(self, src_ip: str, port: int) -> None:
"""Redirect peer's TCP traffic to a local transparent-proxy port."""
self._wg_iptables([
'-t', 'nat', '-A', self.CONNECTIVITY_CHAIN,
'-s', src_ip, '-p', 'tcp',
'-j', 'REDIRECT', '--to-ports', str(self.TOR_TRANS_PORT),
'-j', 'REDIRECT', '--to-ports', str(port),
])
def _add_tor_redirect(self, src_ip: str) -> None:
"""Redirect peer's TCP traffic to local Tor TransPort."""
self._add_redirect(src_ip, self.TOR_TRANS_PORT)
def _add_killswitch(self, mark: int, iface: Optional[str]) -> None:
"""Drop marked packets that would egress via any interface other than iface.
@@ -462,7 +806,13 @@ class ConnectivityManager(BaseServiceManager):
'!', '-o', iface, '-j', 'DROP'])
def _exit_status(self, exit_type: str) -> Dict[str, Any]:
"""Return per-exit status (config presence + interface up/down)."""
"""Return per-exit status (config presence + interface up/down).
An exit counts as configured when a legacy uploaded config file
exists, OR the backing store service is installed, OR its container
is running — exits installed via the Service Store never create the
legacy upload files.
"""
info: Dict[str, Any] = {'configured': False, 'iface_up': False}
if exit_type == 'wireguard_ext':
path = os.path.join(self.wireguard_ext_dir, 'wg_ext0.conf')
@@ -477,6 +827,18 @@ class ConnectivityManager(BaseServiceManager):
info['configured'] = False
elif exit_type == 'tor':
info['configured'] = True # Tor uses defaults; no per-cell config
elif exit_type == 'sshuttle':
info['configured'] = os.path.isfile(
os.path.join(self.sshuttle_dir, 'sshuttle.conf'))
elif exit_type == 'proxy':
info['configured'] = os.path.isfile(
os.path.join(self.proxy_dir, 'redsocks.conf'))
if not info['configured'] and (
self._store_service_installed(exit_type)
or self._exit_container_running(exit_type)
):
info['configured'] = True
iface = self.IFACES.get(exit_type)
if iface:
@@ -485,8 +847,41 @@ class ConnectivityManager(BaseServiceManager):
info['iface_up'] = r.returncode == 0 and 'UP' in (r.stdout or '')
except Exception:
info['iface_up'] = False
if info['iface_up']:
info['status'] = 'active'
elif info['configured']:
info['status'] = 'configured'
else:
info['status'] = 'not_configured'
return info
def _store_service_installed(self, exit_type: str) -> bool:
"""True when the store service backing this exit type is installed."""
svc_id = self.STORE_SERVICE_IDS.get(exit_type)
if not svc_id or self.config_manager is None:
return False
try:
installed = self.config_manager.get_installed_services()
except Exception as e:
logger.debug(f"_store_service_installed({exit_type}): {e}")
return False
return isinstance(installed, dict) and svc_id in installed
def _exit_container_running(self, exit_type: str) -> bool:
"""True when the exit's container is currently running."""
cname = self.EXIT_CONTAINERS.get(exit_type)
if not cname:
return False
try:
r = subprocess.run(
['docker', 'inspect', '-f', '{{.State.Running}}', cname],
capture_output=True, text=True, timeout=5,
)
return r.returncode == 0 and r.stdout.strip() == 'true'
except Exception:
return False
def _peer_source_ip(self, peer_name: str) -> Optional[str]:
"""Return a peer's WireGuard IP (no /CIDR suffix)."""
if not peer_name or self.peer_registry is None: