Gan, Jimmy
44f0ed5d04
Deploy Dashboard (Dev) / deploy-dev (push) Successful in 3m29s
Tighten terminal websocket auth and proxy trust handling while making file-backed auth/RBAC writes atomic to reduce high-impact security and persistence risks. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
import os
|
|
import ipaddress
|
|
from fastapi import Request
|
|
# Dashboard v1.3 — Runner DNS fix applied
|
|
|
|
GITEA_URL = os.environ.get("GITEA_URL", "http://gitea:3000")
|
|
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
|
|
DOCKER_HOST = os.environ.get("DOCKER_HOST", "tcp://docker-socket-proxy:2375")
|
|
VOLUME_ROOT = os.environ.get("VOLUME_ROOT", "/volume1")
|
|
|
|
# Auth
|
|
SECRET_KEY = os.environ.get("SECRET_KEY", "")
|
|
if not SECRET_KEY or len(SECRET_KEY) < 32:
|
|
raise RuntimeError("SECRET_KEY must be set and at least 32 characters")
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
|
REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days
|
|
ADMIN_USER = os.environ.get("ADMIN_USER", "admin")
|
|
ADMIN_PASSWORD_HASH = os.environ.get("ADMIN_PASSWORD_HASH", "")
|
|
TOTP_SECRET = os.environ.get("TOTP_SECRET", "")
|
|
|
|
# SSH terminal
|
|
SSH_HOST = os.environ.get("SSH_HOST", "host.docker.internal")
|
|
SSH_USER = os.environ.get("SSH_USER", "zjgump")
|
|
SSH_KEY_PATH = os.environ.get("SSH_KEY_PATH", "/app/ssh/id_ed25519")
|
|
SSH_KNOWN_HOSTS = os.environ.get("SSH_KNOWN_HOSTS", "/app/ssh/known_hosts")
|
|
|
|
VPS_SSH_HOST = os.environ.get("VPS_SSH_HOST", "158.101.140.85")
|
|
VPS_SSH_USER = os.environ.get("VPS_SSH_USER", "jimmyg")
|
|
VPS_SSH_KEY_PATH = os.environ.get("VPS_SSH_KEY_PATH", "/app/ssh/vps_id_ed25519")
|
|
|
|
CADDY_VPS_SSH_HOST = os.environ.get("CADDY_VPS_SSH_HOST", "161.33.182.109")
|
|
CADDY_VPS_SSH_USER = os.environ.get("CADDY_VPS_SSH_USER", "ubuntu")
|
|
|
|
MAC_SSH_HOST = os.environ.get("MAC_SSH_HOST", "192.168.31.22")
|
|
MAC_SSH_USER = os.environ.get("MAC_SSH_USER", "jimmyg")
|
|
MAC_SSH_KEY_PATH = os.environ.get("MAC_SSH_KEY_PATH", "/app/ssh/mac_id_ed25519")
|
|
|
|
# Telegram notifications
|
|
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
|
|
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
|
|
TELEGRAM_PROXY = os.environ.get("TELEGRAM_PROXY", "")
|
|
|
|
WEBAUTHN_RP_ID = os.environ.get("WEBAUTHN_RP_ID", "nas.jimmygan.com")
|
|
WEBAUTHN_ORIGINS = os.environ.get("WEBAUTHN_ORIGIN", "https://nas.jimmygan.com,https://nas.jimmygan.com:8443").split(",")
|
|
|
|
# CORS
|
|
CORS_ORIGINS = os.environ.get("CORS_ORIGINS", "https://nas.jimmygan.com,https://nas.jimmygan.com:8443").split(",")
|
|
|
|
# Chat Summary
|
|
CHAT_SUMMARY_DB = os.environ.get("CHAT_SUMMARY_DB", "/app/data/chat-summarizer/chat_summary.db")
|
|
|
|
# Info Engine
|
|
INFO_ENGINE_DB = os.environ.get("INFO_ENGINE_DB", "/app/data/info-engine/info_engine.db")
|
|
|
|
# LiteLLM
|
|
LITELLM_URL = os.environ.get("LITELLM_URL", "http://127.0.0.1:4005")
|
|
LITELLM_HEALTH_API_KEY = os.environ.get("LITELLM_HEALTH_API_KEY", "")
|
|
|
|
# CC Connect
|
|
CC_CONNECT_URL = os.environ.get("CC_CONNECT_URL", "")
|
|
|
|
# OpenClaw
|
|
OPENCLAW_GATEWAY_TOKEN = os.environ.get("OPENCLAW_GATEWAY_TOKEN", "")
|
|
|
|
# Networking configs
|
|
LAN_IP_RANGES = os.environ.get("LAN_IP_RANGES", "192.168.31.0/24").split(",")
|
|
TRUSTED_PROXIES = os.environ.get("TRUSTED_PROXIES", "127.0.0.1,::1,172.21.0.1").split(",")
|
|
|
|
|
|
def _parse_ip(ip_str: str):
|
|
try:
|
|
return ipaddress.ip_address((ip_str or "").strip())
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _ip_matches_ranges(ip_str: str, ranges: list[str]) -> bool:
|
|
ip_obj = _parse_ip(ip_str)
|
|
if ip_obj is None:
|
|
return False
|
|
|
|
for subnet_str in ranges:
|
|
subnet_str = subnet_str.strip()
|
|
if not subnet_str:
|
|
continue
|
|
try:
|
|
if "/" in subnet_str:
|
|
if ip_obj in ipaddress.ip_network(subnet_str, strict=False):
|
|
return True
|
|
elif ip_obj == ipaddress.ip_address(subnet_str):
|
|
return True
|
|
except ValueError:
|
|
continue
|
|
return False
|
|
|
|
|
|
def is_lan_ip(ip_str: str) -> bool:
|
|
"""Check if IP is in local LAN (not Tailscale)"""
|
|
return _ip_matches_ranges(ip_str, LAN_IP_RANGES)
|
|
|
|
|
|
def is_tailscale_ip(ip_str: str) -> bool:
|
|
"""Check if IP is in Tailscale CGNAT range (100.64.0.0/10)"""
|
|
return _ip_matches_ranges(ip_str, ["100.64.0.0/10"])
|
|
|
|
|
|
def is_trusted_proxy(ip_str: str) -> bool:
|
|
return _ip_matches_ranges(ip_str, TRUSTED_PROXIES)
|
|
|
|
|
|
def get_forwarded_client_ip(request: Request) -> str | None:
|
|
source_ip = request.client.host if request.client else ""
|
|
if not is_trusted_proxy(source_ip):
|
|
return None
|
|
|
|
forwarded_for = request.headers.get("x-forwarded-for", "")
|
|
forwarded_ip = forwarded_for.split(",", 1)[0].strip()
|
|
return forwarded_ip or None
|
|
|
|
|
|
def get_client_ip(request: Request) -> str:
|
|
return get_forwarded_client_ip(request) or (request.client.host if request.client else "")
|