app.py
Rama main · réplica de lectura del repositorio original.
import fcntl
import copy
import os
import re
import time
import base64
import hashlib
import hmac
from urllib.parse import quote, urlparse
import shutil
import secrets
import threading
import ipaddress
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
from datetime import datetime, timezone, timedelta
import click
from flask import (Flask, render_template, request, redirect,
url_for, flash, jsonify, abort, session, send_file, g, has_request_context)
from functools import wraps
from werkzeug.middleware.proxy_fix import ProxyFix
from ruamel.yaml import YAML
from ruamel.yaml import YAML as SafeYAML
from ruamel.yaml.scalarstring import DoubleQuotedScalarString
from io import StringIO
from core import env
from core.env import GITHUB_REPO, APP_VERSION, logger, PROXY_FIX_HOPS
from core import crypto
from core import config as _cfg
yaml = _cfg.yaml
_yaml_safe = _cfg.yaml_safe
_safe_file_path = _cfg.safe_file_path
_readable_config_path = _cfg.readable_config_path
_is_safe_path = _cfg.is_safe_path
_resolve_config_path = _cfg.resolve_config_path
_safe_api_url = _cfg.safe_api_url
_same_api_origin = _cfg.same_api_origin
_sanitize_go_templates = _cfg.sanitize_go_templates
_restore_go_templates = _cfg.restore_go_templates
load_config = _cfg.load_config
_strip_empty_sections = _cfg.strip_empty_sections
save_config = _cfg.save_config
_svc_key = _cfg.svc_key
_as_dict = _cfg.as_dict
_load_config_display = _cfg._load_config_display
_get_config_parse_errors = _cfg._get_config_parse_errors
from core import agents_store as _ag
from core import settings as _settings
OPTIONAL_TABS = _settings.OPTIONAL_TABS
load_settings = _settings.load_settings
save_settings = _settings.save_settings
update_settings = _settings.update_settings
_get_acme_json_path = _settings._get_acme_json_path
_get_access_log_path = _settings._get_access_log_path
_get_static_config_path = _settings._get_static_config_path
_get_restart_method = _settings._get_restart_method
load_agents = _ag.load_agents
save_agents_file = _ag.save_agents_file
_save_agents = _ag.encrypt_agents
_parse_agent_dict = _ag.parse_agent_dict
from core import backups as _back
from core import notifications as _noti
from core import notify_providers as _notify_providers
from core import monitor as _monitor
from core import reachability as _reach
from core import names as _naming
from core import providers as _providers
from core import cert_usage as _cert_usage
from core import locks as _locks
from core import acme_store as _acme
from core import route_health as _rh
from core import oidc_tokens as _oidc_tokens
from core import updates as _updates
from core import traefik as _trae
from core import agents_http as _agen
from core import agent_errors as _agent_err
from core import git as _git
from core import auth as _auth
from core import routes_build as _rb
from flask_babel import gettext
from core import i18n as _i18n
from core import crowdsec as _crowd
from core import certs as _certs
_parse_cert_expiry = _certs._parse_cert_expiry
_certs_from_tls_configs = _certs._certs_from_tls_configs
from core import self_route as _self_r
_self_route_path = _self_r._self_route_path
_detect_self_route_domain = _self_r._detect_self_route_domain
_detect_self_route_from_own_labels = _self_r._detect_self_route_from_own_labels
_find_existing_self_route = _self_r._find_existing_self_route
_write_self_route = _self_r._write_self_route
_delete_self_route = _self_r._delete_self_route
SELF_ROUTE_FILENAME = _self_r.SELF_ROUTE_FILENAME
_cs_lapi_url = _crowd._cs_lapi_url
_cs_api_key = _crowd._cs_api_key
_cs_machine_id = _crowd._cs_machine_id
_cs_machine_password = _crowd._cs_machine_password
_cs_has_machine = _crowd._cs_has_machine
_cs_has_cert = _crowd._cs_has_cert
_cs_tls_kwargs = _crowd._cs_tls_kwargs
cs_timeout = _crowd.cs_timeout
cs_alert_limit = _crowd.cs_alert_limit
_cs_request = _crowd._cs_request
_cs_request_strict = _crowd._cs_request_strict
CrowdSecUnavailable = _crowd.CrowdSecUnavailable
_cs_jwt_cache = _crowd._cs_jwt_cache
_cs_jwt = _crowd._cs_jwt
_cs_machine_request = _crowd._cs_machine_request
from core import geoip as _geoip
_geoip_enabled = _geoip._geoip_enabled
_geoip_db_path = _geoip._geoip_db_path
_geoip_reader = _geoip._geoip_reader
_geoip_lookup = _geoip._geoip_lookup
_geoip_status = _geoip._geoip_status
_geoip_download = _geoip._geoip_download
_DBIP_URL = _geoip._DBIP_URL
_GEOIP_SENTINEL = _geoip._GEOIP_SENTINEL
_geoip_cache = _geoip._geoip_cache
_geoip_lock = _geoip._geoip_lock
_geoip_state = _geoip._geoip_state
_trusted_ip_key = _rb._trusted_ip_key
_merge_trusted_ips = _rb._merge_trusted_ips
_apply_managed_keys = _rb._apply_managed_keys
_merge_router = _rb._merge_router
_merge_service = _rb._merge_service
from core import composite_services as _composite
from core import service_ownership as _svc_own
def _router_resolves_to_composite(router_name: str, agent=None) -> bool:
if not router_name:
return False
try:
configs = list(_agent_load_configs(agent).values()) if agent else [
load_config(p) for p in env.CONFIG_PATHS]
except Exception:
return False
for cfg in configs:
for section in ('http', 'tcp', 'udp'):
routers = (cfg.get(section) or {}).get('routers') or {}
router = routers.get(router_name)
if not isinstance(router, dict):
continue
svc = str(router.get('service') or '').split('@')[0]
svc_def = ((cfg.get(section) or {}).get('services') or {}).get(svc)
if _svc_own.composite_type(svc_def):
return True
return False
def _svc_ledger_key(name, agent_id=''):
return _svc_own.ledger_key(name, agent_id)
_json_plain = _rb._json_plain
_headers_preset_defaults = _rb._headers_preset_defaults
_build_permissions_policy = _rb._build_permissions_policy
_build_headers_middleware = _rb._build_headers_middleware
_parse_permissions_policy = _rb._parse_permissions_policy
_decode_headers_middleware = _rb._decode_headers_middleware
_to_list = _rb._to_list
_service_type = _rb._service_type
_build_apps = _rb._build_apps
_build_middlewares = _rb._build_middlewares
_traefik_router_ep_map = _rb._traefik_router_ep_map
_traefik_service_url_map = _rb._traefik_service_url_map
_build_external_routes = _rb._build_external_routes
_entrypoint_mw_map = _rb._entrypoint_mw_map
_build_all_apps = _rb._build_all_apps
HEADERS_PRESET_FEATURES = _rb.HEADERS_PRESET_FEATURES
HEADERS_PRESET_HSTS_SECONDS = _rb.HEADERS_PRESET_HSTS_SECONDS
HEADERS_PRESET_REFERRER_DEFAULT = _rb.HEADERS_PRESET_REFERRER_DEFAULT
HEADERS_PRESET_REFERRER_VALUES = _rb.HEADERS_PRESET_REFERRER_VALUES
HEADERS_PRESET_SELF_DEFAULT = _rb.HEADERS_PRESET_SELF_DEFAULT
_HEADERS_PRESET_KEYS = _rb._HEADERS_PRESET_KEYS
_PERM_TOKEN_TO_VALUE = _rb._PERM_TOKEN_TO_VALUE
_PERM_VALUE_TO_TOKEN = _rb._PERM_VALUE_TO_TOKEN
_auth_enabled = _auth._auth_enabled
_oidc_active = _auth._oidc_active
_auth_required = _auth._auth_required
_get_csrf_token = _auth._get_csrf_token
_check_csrf = _auth._check_csrf
csrf_protect = _auth.csrf_protect
_check_password = _auth._check_password
_verify_api_key = _auth._verify_api_key
_is_authenticated = _auth._is_authenticated
_check_inactivity = _auth._check_inactivity
_check_api_key = _auth._check_api_key
login_required = _auth.login_required
_CsrfError = _auth._CsrfError
_git_repo_dir = _git._git_repo_dir
_valid_git_url = _git._valid_git_url
_safe_git_branch = _git._safe_git_branch
_git_askpass_path = _git._git_askpass_path
_git_run = _git._git_run
_git_ensure_repo_at = _git._git_ensure_repo_at
_git_ensure_repo = _git._git_ensure_repo
_git_lock = _git._git_lock
_git_push_configs = _git._git_push_configs
_git_push_if_enabled = _git._git_push_if_enabled
_git_agent_repo_dir = _git._git_agent_repo_dir
_agent_git_branch = _git._agent_git_branch
_git_push_agent_configs = _git._git_push_agent_configs
_git_push_agent_if_enabled = _git._git_push_agent_if_enabled
_git_show_first = _git._git_show_first
_GIT_ALLOWED_SCHEMES = _git._GIT_ALLOWED_SCHEMES
_GIT_PROTO_HARDENING = _git._GIT_PROTO_HARDENING
ensure_backup_dir = _back.ensure_backup_dir
_backup_keep_count = _back._backup_keep_count
_prune_backups = _back._prune_backups
create_backup = _back.create_backup
_send_webhook = _noti._send_webhook
_fire_webhook = _noti._fire_webhook
_load_notifications = _noti._load_notifications
get_notifications = _noti.get_notifications
delete_notification = _noti.delete_notification
clear_notifications = _noti.clear_notifications
add_notification = _noti.add_notification
_traefik_verify = _trae._traefik_verify
traefik_api_get = _trae.traefik_api_get
traefik_api_get_all = _trae.traefik_api_get_all
_fetch_traefik_routers_and_services = _trae._fetch_traefik_routers_and_services
_agent_by_id = _agen._agent_by_id
_agent_request = _agen._agent_request
_agent_load_configs = _agen._agent_load_configs
_agent_write_config = _agen._agent_write_config
_notifications = _noti._notifications
_notif_lock = _noti._notif_lock
class _BasePathMiddleware:
def __init__(self, wsgi_app, prefix):
self.wsgi_app = wsgi_app
self.prefix = prefix
def __call__(self, environ, start_response):
path = environ.get('PATH_INFO', '')
if path == self.prefix or path.startswith(self.prefix + '/'):
environ['PATH_INFO'] = path[len(self.prefix):] or '/'
environ['SCRIPT_NAME'] = self.prefix
return self.wsgi_app(environ, start_response)
class _TrustedProxyFix:
def __init__(self, wsgi_app, hops):
self.raw = wsgi_app
self.fixed = ProxyFix(wsgi_app, x_for=hops, x_proto=1, x_host=1)
def __call__(self, environ, start_response):
if env.peer_is_trusted(environ.get('REMOTE_ADDR')):
return self.fixed(environ, start_response)
return self.raw(environ, start_response)
app = Flask(__name__)
app.wsgi_app = _i18n.LocalePrefixMiddleware(app.wsgi_app)
app.wsgi_app = _TrustedProxyFix(app.wsgi_app, PROXY_FIX_HOPS)
if env.BASE_PATH:
app.wsgi_app = _BasePathMiddleware(app.wsgi_app, env.BASE_PATH)
app.config['APPLICATION_ROOT'] = env.BASE_PATH
_CONFIG_DIR = os.path.dirname(os.environ.get('SETTINGS_PATH', '/app/config/manager.yml'))
_SECRET_KEY_PATH = os.path.join(_CONFIG_DIR, '.secret_key')
def _load_or_create_secret_key() -> bytes:
env_key = os.environ.get('SECRET_KEY', '').strip()
if env_key:
if len(env_key) < 32:
raise SystemExit("SECRET_KEY must be at least 32 characters. Generate one with: python3 -c \"import secrets; print(secrets.token_hex(32))\"")
return env_key.encode()
if os.path.exists(_SECRET_KEY_PATH):
key = open(_SECRET_KEY_PATH, 'rb').read().strip()
if len(key) >= 32:
return key
key = secrets.token_hex(32).encode()
key_dir = os.path.dirname(_SECRET_KEY_PATH)
os.makedirs(key_dir, exist_ok=True)
tmp = os.path.join(key_dir, '.secret_key.%d.tmp' % os.getpid())
with open(tmp, 'wb') as f:
f.write(key)
try:
os.chmod(tmp, 0o600)
except OSError:
pass
try:
os.link(tmp, _SECRET_KEY_PATH)
except FileExistsError:
existing = open(_SECRET_KEY_PATH, 'rb').read().strip()
if len(existing) >= 32:
key = existing
except OSError:
with open(_SECRET_KEY_PATH, 'wb') as f:
f.write(key)
finally:
try:
os.unlink(tmp)
except OSError:
pass
return key
app.secret_key = _load_or_create_secret_key()
_OTP_KEY_PATH = env.OTP_KEY_PATH
_get_otp_fernet = crypto.get_otp_fernet
_encrypt_otp_secret = crypto.encrypt_secret
_decrypt_otp_secret = crypto.decrypt_secret
def load_templates() -> list:
if not os.path.exists(TEMPLATES_PATH):
return []
try:
with open(TEMPLATES_PATH, 'r') as f:
raw = _yaml_safe.load(f) or {}
return [
{'id': str(t['id']), 'name': str(t.get('name', ''))[:100], 'yaml': str(t.get('yaml', ''))}
for t in (raw.get('templates', []) or [])
if isinstance(t, dict) and t.get('id') and t.get('name')
]
except Exception as e:
logger.warning(f"Could not load templates.yml: {e}")
return []
def save_templates_file(templates: list):
import json as _json
os.makedirs(os.path.dirname(TEMPLATES_PATH), exist_ok=True)
tmp = f"{TEMPLATES_PATH}.tmp.{os.getpid()}.{threading.get_ident()}"
safe = [{'id': t['id'], 'name': t['name'], 'yaml': t['yaml']} for t in templates]
try:
with open(tmp, 'w') as f:
yaml.dump({'templates': _json.loads(_json.dumps(safe))}, f)
os.replace(tmp, TEMPLATES_PATH)
finally:
if os.path.exists(tmp):
try:
os.unlink(tmp)
except OSError:
pass
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7)
app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
app.config['SESSION_COOKIE_SECURE'] = os.environ.get('COOKIE_SECURE', 'false').lower() == 'true'
INACTIVITY_TIMEOUT = _auth.INACTIVITY_TIMEOUT
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from core import rate_store as _rate_store
limiter = Limiter(get_remote_address, app=app, default_limits=[], storage_uri=_rate_store.SCHEME)
def _failed_sign_in(response):
return response.status_code == 200
_login_failure_limit = limiter.shared_limit(
lambda: env.LOGIN_FAILURE_LIMIT or env.DEFAULT_LOGIN_FAILURE_LIMIT, scope='login-failures',
key_func=lambda: 'account', methods=['POST'], deduct_when=_failed_sign_in,
exempt_when=lambda: not env.LOGIN_FAILURE_LIMIT,
error_message=lambda: gettext('Too many failed sign-in attempts. Try again later.'))
_otp_failure_limit = limiter.shared_limit(
lambda: env.OTP_FAILURE_LIMIT or env.DEFAULT_OTP_FAILURE_LIMIT, scope='otp-failures',
key_func=lambda: 'account', methods=['POST'], deduct_when=_failed_sign_in,
exempt_when=lambda: not env.OTP_FAILURE_LIMIT,
error_message=lambda: gettext('Too many failed two-factor codes. Try again later.'))
def _setup_checks_a_code() -> bool:
if request.method != 'POST':
return False
try:
s = load_settings()
except Exception:
return False
return bool(s.get('setup_password_reset') and s.get('otp_enabled') and s.get('otp_secret'))
_setup_otp_failure_limit = limiter.shared_limit(
lambda: env.OTP_FAILURE_LIMIT or env.DEFAULT_OTP_FAILURE_LIMIT, scope='otp-failures',
key_func=lambda: 'account', methods=['POST'], deduct_when=_failed_sign_in,
exempt_when=lambda: not env.OTP_FAILURE_LIMIT or not _setup_checks_a_code(),
error_message=lambda: gettext('Too many failed two-factor codes. Try again later.'))
BACKUP_DIR = env.BACKUP_DIR
SETTINGS_PATH = env.SETTINGS_PATH
_CONFIG_DIR = env.CONFIG_DIR
GROUPS_CACHE_DIR = env.GROUPS_CACHE_DIR
GEOIP_DIR = env.GEOIP_DIR
GROUPS_CONFIG_FILE = env.GROUPS_CONFIG_FILE
NOTIFICATIONS_PATH = env.NOTIFICATIONS_PATH
AGENTS_PATH = env.AGENTS_PATH
TEMPLATES_PATH = env.TEMPLATES_PATH
ACTIVE_CONFIG_DIR = env.ACTIVE_CONFIG_DIR
_ALLOWED_API_SCHEMES = env.ALLOWED_API_SCHEMES
def _ssrf_ok(url: str) -> bool:
return _reach.ssrf_ok(url)
_EDIT_DICTS = ('managed_middlewares', 'disabled_routes')
def _edit_scope() -> str:
agent_id = (request.form.get('agent_id', '') or request.args.get('agent_id', '')
or request.args.get('server', ''))
if not agent_id:
body = request.get_json(force=True, silent=True)
if isinstance(body, dict):
agent_id = body.get('agent_id') or body.get('server') or ''
agent_id = str(agent_id or '').strip()
return f'config:agent:{agent_id}' if agent_id else 'config:local'
def _config_edit(fn):
@wraps(fn)
def inner(*args, **kwargs):
with _locks.config_edit_lock(_edit_scope()):
owner = 'edit_base' not in g
if owner:
current = load_settings()
g.edit_base = {k: copy.deepcopy(current.get(k) or {}) for k in _EDIT_DICTS}
try:
return fn(*args, **kwargs)
finally:
if owner:
g.pop('edit_base', None)
return inner
def _save_edit_dicts(**after):
base = g.get('edit_base') if has_request_context() else None
if base is None:
return update_settings(**after)
_settings.merge_settings_dicts({k: base.get(k) or {} for k in after}, **after)
for k, v in after.items():
base[k] = copy.deepcopy(v or {})
def _agents_locked(fn):
@wraps(fn)
def inner(*args, **kwargs):
with _locks.file_lock(env.AGENTS_PATH):
return fn(*args, **kwargs)
return inner
def _register_config_path(path: str):
env.register_config_path(path)
GEOIP_CHECK_INTERVAL = 86400
def _geoip_autoupdate_loop():
lock_path = os.path.join(env.CONFIG_DIR, '.geoip.lock')
while True:
fh = None
try:
fh = open(lock_path, 'a+')
fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
_geoip_maybe_autoupdate()
except (OSError, BlockingIOError):
pass
except Exception:
logger.exception("GeoIP auto-update check failed")
finally:
if fh is not None:
try:
fh.close()
except Exception:
pass
time.sleep(GEOIP_CHECK_INTERVAL)
def _geoip_maybe_autoupdate():
try:
if not _geoip_enabled():
return
path = _geoip_db_path()
stale = True
if os.path.exists(path):
try:
stale = (time.time() - os.path.getmtime(path)) > 35 * 86400
except OSError:
stale = True
if stale:
_geoip_download()
except Exception:
logger.exception("GeoIP auto-update failed")
def _get_traefik_container() -> str:
return os.environ.get('TRAEFIK_CONTAINER', 'traefik')
def _get_signal_file_path() -> str:
return os.environ.get('SIGNAL_FILE_PATH', '/signals/restart.sig')
def _best_entrypoint() -> str:
eps = traefik_api_get_all('/api/entrypoints') or []
for ep in eps:
addr = ep.get('address', '')
if ':443' in addr or '/443' in addr:
return ep.get('name', 'websecure')
if eps:
return eps[0].get('name', 'websecure')
return 'websecure'
def _detect_provider_tabs() -> list:
try:
return sorted(_providers.tabs_from_overview(traefik_api_get('/api/overview')))
except Exception:
logger.debug("Could not read the Traefik providers for the setup wizard")
return []
def _detect_setup_self_route() -> tuple[str, str]:
settings = load_settings()
saved = settings.get('self_route', {})
if saved.get('domain'):
return saved['domain'], saved.get('service_url', 'http://traefik-manager:5000')
domain = _detect_self_route_domain()
if domain:
return domain, 'http://traefik-manager:5000'
return _detect_self_route_from_own_labels()
def _password_error(pw: str, new: bool = False) -> str | None:
if len(pw) < 8:
if new:
return gettext('New password must be at least 8 characters.')
return gettext('Password must be at least 8 characters.')
if len(pw.encode('utf-8')) > 72:
if new:
return gettext('New password must be 72 bytes or fewer, which is the bcrypt limit. '
'Accented and non-Latin characters take more than one byte each.')
return gettext('Password must be 72 bytes or fewer, which is the bcrypt limit. '
'Accented and non-Latin characters take more than one byte each.')
return None
def _hash_password(plaintext: str) -> str:
import bcrypt
return bcrypt.hashpw(plaintext.encode(), bcrypt.gensalt(rounds=12)).decode()
def _ensure_password():
if os.environ.get('ADMIN_PASSWORD', '').strip():
return
if load_settings().get('password_hash', ''):
return
def generate(current):
if current.get('password_hash', ''):
return None
password = secrets.token_urlsafe(16)
logger.warning("=" * 60)
logger.warning(" TRAEFIK MANAGER - AUTO-GENERATED PASSWORD")
logger.warning(f" Password: {password}")
logger.warning(" Log in with this password, complete setup, then")
logger.warning(" you will be prompted to set a permanent password.")
logger.warning("=" * 60)
return {
'password_hash': _hash_password(password),
'must_change_password': True,
'setup_complete': False,
}
_settings.modify_settings(generate)
def _sync_admin_password_fingerprint():
admin_pw = os.environ.get('ADMIN_PASSWORD', '').strip()
if not admin_pw:
return
key = app.secret_key if isinstance(app.secret_key, bytes) else str(app.secret_key).encode()
fingerprint = hmac.new(key, admin_pw.encode(), hashlib.sha256).hexdigest()
try:
stored = load_settings().get('admin_password_fp', '')
if stored and hmac.compare_digest(stored, fingerprint):
return
if stored:
_settings.bump_session_epoch(admin_password_fp=fingerprint)
logger.warning("ADMIN_PASSWORD changed since the last start, so every browser session was signed out")
else:
update_settings(admin_password_fp=fingerprint)
except Exception:
logger.exception("Could not record the ADMIN_PASSWORD fingerprint")
def _read_traefik_labels():
try:
import docker as _docker
client = _docker.from_env()
container_name = os.environ.get('TRAEFIK_CONTAINER', 'traefik')
for c in client.containers.list():
labels = c.labels or {}
if labels.get('traefik-manager.role') == 'traefik' or c.name == container_name:
if not os.environ.get('RESTART_METHOD') and labels.get('traefik-manager.restart-method'):
os.environ['RESTART_METHOD'] = labels['traefik-manager.restart-method']
if not os.environ.get('STATIC_CONFIG_PATH') and labels.get('traefik-manager.static-config'):
os.environ['STATIC_CONFIG_PATH'] = labels['traefik-manager.static-config']
if not os.environ.get('TRAEFIK_CONTAINER'):
os.environ['TRAEFIK_CONTAINER'] = c.name
logger.info(f"Static config: read labels from Traefik container {c.name!r}")
break
except Exception:
pass
_read_traefik_labels()
_s = load_settings()
_static_path = _get_static_config_path()
_restart_meth = _get_restart_method()
_oidc_on = bool(_s.get('oidc_issuer'))
logger.info("===========================================")
logger.info(f"Traefik Manager v{APP_VERSION}")
if env.MULTI_CONFIG:
for _cp in env.CONFIG_PATHS:
logger.info(f"Config File: {_cp}")
elif ACTIVE_CONFIG_DIR:
logger.info(f"Config Dir: {ACTIVE_CONFIG_DIR}")
else:
logger.info(f"Config Path: {env.CONFIG_PATH}")
logger.info(f"Settings Path: {SETTINGS_PATH}")
logger.info(f"Backup Dir: {BACKUP_DIR}")
logger.info(f"Traefik API: {_s['traefik_api_url']}")
logger.info(f"Restart Method: {_restart_meth}")
logger.info(f"Trusted Hops: {PROXY_FIX_HOPS}")
logger.info(f"Trusted Proxies: {', '.join(env.trusted_proxies_list())}")
logger.info(f"Static Config: {_static_path if _static_path else 'not configured'}")
logger.info(f"Domains: {_s['domains']}")
logger.info(f"Cert Resolver: {_s['cert_resolver'] or 'not set'}")
logger.info(f"Auth Enabled: {_auth_enabled()}")
if _s.get('oidc_enabled'):
logger.info(f"OIDC: enabled ({_s.get('oidc_issuer', '')})")
if not _auth_enabled() and not _s.get('oidc_enabled'):
if _s.get('auth_external_ack'):
logger.info("Auth: delegated to an external provider (acknowledged in Settings)")
else:
logger.warning("SECURITY: no authentication is active - the web UI is publicly accessible. Enable a password or OIDC.")
if _s.get('oidc_enabled') and not _s.get('oidc_allowed_emails', '').strip() and not _s.get('oidc_allowed_groups', '').strip() and not _s.get('oidc_allow_any_authenticated'):
logger.warning("SECURITY: OIDC is enabled with no allowed emails/groups - logins are denied until you set an allowlist or enable 'Allow any authenticated account'.")
logger.info("===========================================")
_ensure_password()
_sync_admin_password_fingerprint()
_i18n.init_app(app, lambda: load_settings().get('default_language', ''))
@app.context_processor
def _inject_theme():
try:
return {'default_theme': load_settings().get('default_theme', 'dark')}
except Exception:
return {'default_theme': 'dark'}
@app.errorhandler(_CsrfError)
def _handle_csrf_error(e):
return jsonify({'ok': False, 'message': gettext('Session expired - please refresh the page.')}), 403
@app.errorhandler(401)
def _handle_unauthorized(e):
if request.path.startswith('/api/'):
return jsonify({'ok': False, 'error': gettext('Not authenticated'), 'auth_required': True}), 401
return redirect(url_for('login', next=request.path))
@app.context_processor
def inject_csrf():
return {'csrf_token': _get_csrf_token()}
def _static_build_stamp() -> str:
newest = 0.0
static_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'static')
for sub in ('css', 'js'):
root = os.path.join(static_dir, sub)
for dirpath, _dirs, files in os.walk(root):
for name in files:
try:
newest = max(newest, os.path.getmtime(os.path.join(dirpath, name)))
except OSError:
pass
return str(int(newest))
ASSET_VERSION = f"{APP_VERSION}-{_static_build_stamp()}"
@app.context_processor
def inject_asset_version():
return {'asset_version': ASSET_VERSION}
@app.context_processor
def inject_base_path():
return {'base_path': env.BASE_PATH}
def _hash_api_key(key: str) -> str:
import hashlib
return 'sha256:' + hashlib.sha256(key.encode()).hexdigest()
_CONTROL_CHARS_RE = re.compile(r'[\x00-\x1f\x7f]')
def _safe_next(next_url: str) -> str:
nu = _CONTROL_CHARS_RE.sub('', (next_url or '')).strip()
if nu.startswith('/') and not nu.startswith('//') and not nu.startswith('/\\'):
root = request.script_root
if root and nu != root and not nu.startswith(root + '/'):
return root + nu
return nu
return url_for('index')
def _has_password_set() -> bool:
if os.environ.get('ADMIN_PASSWORD', '').strip():
return True
if _settings.settings_unreadable():
return True
return bool(load_settings().get('password_hash', ''))
def _get_effective_hash() -> str:
admin_pw = os.environ.get('ADMIN_PASSWORD', '').strip()
if admin_pw:
return _hash_password(admin_pw)
return load_settings().get('password_hash', '')
_load_notifications()
threading.Thread(target=_geoip_autoupdate_loop, daemon=True).start()
_monitor.register('crowdsec', _crowd.CS_ALERT_INTERVAL,
lambda: _crowd.check_local_alerts(_crowd.CS_ALERT_WINDOW))
_monitor.register('updates', _updates.UPDATE_INTERVAL, _updates.check_updates)
_monitor.register('notify-flush', _noti.FLUSH_INTERVAL, _noti.flush_due)
def _route_health_sources():
out = []
try:
apps, _mws = _build_all_apps(include_external=True, include_internal=True)
out.append((_monitor.HOST_SERVER, '', apps,
{'http': traefik_api_get_all('/api/http/services') or []}))
except Exception:
logger.exception("Route check could not list the host routes")
for server, name, agent in _monitor._agent_servers():
try:
if not _monitor._agent_usable(agent):
continue
payload = _agent_routes_payload(agent, server)
out.append((server, name, payload['apps'], payload.get('traefikServices') or {}))
except Exception:
logger.exception(f"Route check could not list routes for agent {name}")
return out
_monitor.register('routes', _rh.TICK, lambda: _rh.check(
_route_health_sources, overrides_for=lambda server: _read_groups_config(server)['route_overrides']))
def _reencrypt_file(name, read, write):
crypto.clear_plaintext_seen()
content = read()
if not crypto.plaintext_secrets_seen():
return False
crypto.clear_plaintext_seen()
try:
write(content)
except Exception:
logger.exception(f"Could not re-encrypt the secrets in {name}")
return False
logger.info(f"Secrets written in plain text were re-encrypted in {name}")
return True
def _reencrypt_plaintext_secrets():
rewritten = []
if _reencrypt_file('manager.yml', lambda: load_settings(fresh=True), lambda s: save_settings(
domains=s['domains'], cert_resolver=s['cert_resolver'],
traefik_api_url=s['traefik_api_url'], auth_enabled=s['auth_enabled'],
password_hash=s['password_hash'], visible_tabs=s['visible_tabs'])):
rewritten.append('manager.yml')
if _reencrypt_file('agents.yml', lambda: _ag.load_agents(fresh=True), _ag.save_agents_file):
rewritten.append('agents.yml')
crypto.clear_plaintext_seen()
return rewritten
_reencrypted = _reencrypt_plaintext_secrets()
if _reencrypted:
add_notification(
'warning',
_i18n.lazy_gettext('A secret was written to %(files)s in plain text. It has been encrypted '
'in place and is no longer stored in the clear',
files=', '.join(_reencrypted)),
category='security')
for _label, _path, _err in env.unwritable_storage():
logger.error(f"{_label} storage at {_path} is not writable ({_err}). "
f"Settings, backups and scheduled checks will not survive a restart. "
f"Check the volume or bind mount for this path.")
_cfg.tighten_secret_files(env.SETTINGS_PATH, env.AGENTS_PATH, env.OTP_KEY_PATH)
if not os.environ.get('TRUSTED_PROXIES', '').strip():
logger.info(
"TRUSTED_PROXIES is not set, so forwarding headers are accepted from %s. Any host in "
"those ranges - another container on the same docker network, or any machine on the "
"LAN - can choose the client IP used for rate limits and the audit log, and the host "
"used to build the OIDC redirect_uri. Set TRUSTED_PROXIES to your proxy's address, "
"and OIDC_REDIRECT_URI to the URL registered with your provider.",
', '.join(env.trusted_proxies_list()))
_monitor.start()
_SILENT_PREFIXES = (
'/static/',
'/api/notifications',
'/api/traefik/',
'/api/routes',
'/api/dashboard/',
'/api/manager/version',
'/api/configs',
'/api/settings/tabs',
'/api/ping',
)
@app.before_request
def log_request_info():
path = request.path
method = request.method
if method == 'GET':
if request.remote_addr == '127.0.0.1':
return
if any(path.startswith(p) for p in _SILENT_PREFIXES):
logger.debug(f"{request.remote_addr} → {method} {path}")
return
logger.info(f"{request.remote_addr} → {method} {path}")
@app.after_request
def set_security_headers(response):
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin'
response.headers['X-XSS-Protection'] = '1; mode=block'
response.headers['Permissions-Policy'] = 'camera=(), microphone=(), geolocation=()'
if not request.path.startswith('/static/') and 'Cache-Control' not in response.headers:
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'
if app.config.get('SESSION_COOKIE_SECURE'):
response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
return response
def _close_reset_window(settings):
if not settings.get('setup_password_reset'):
return
update_settings(setup_password_reset=False)
logger.warning("Password reset window closed after a successful sign-in")
def _start_session(remember=False, extra=None, notify=True, note=None):
values = {'authenticated': True,
'last_active': time.time(),
'login_time': datetime.now(timezone.utc).isoformat()}
values.update(extra or {})
session.clear()
session.update(values)
session.permanent = bool(remember)
_auth._stamp_session()
if notify:
add_notification('info', note or _i18n.lazy_gettext('Login from %(ip)s', ip=request.remote_addr),
category='security')
_close_reset_window(load_settings())
_PASSWORD_CHANGE_EXEMPT = {
'static', 'login', 'login_otp', 'logout', 'setup', 'force_change_password',
'api_change_password', 'api_health', 'oidc_login', 'oidc_callback',
'setup_test_git', 'setup_test_crowdsec', 'api_setup_test_connection',
}
@app.before_request
def _require_password_change():
if request.endpoint is None or request.endpoint in _PASSWORD_CHANGE_EXEMPT:
return None
if request.headers.get('X-Api-Key') and _check_api_key():
return None
_auth._drop_stale_session()
if not session.get('authenticated') or os.environ.get('ADMIN_PASSWORD', '').strip():
return None
if session.get('auth_method') == 'oidc' or not _auth_enabled():
return None
settings = load_settings()
if not settings.get('must_change_password'):
return None
if request.path.startswith('/api/'):
return jsonify({'error': gettext('password change required')}), 403
return redirect(url_for('force_change_password') if settings.get('setup_complete') else url_for('setup'))
@app.route('/login', methods=['GET', 'POST'])
@limiter.limit("5 per minute", methods=["POST"])
@_login_failure_limit
def login():
if not _auth_required():
return redirect(url_for('index'))
if session.get('authenticated'):
return redirect(url_for('index'))
settings = load_settings()
local_auth = _auth_enabled()
temp_password_hint = (
settings.get('must_change_password', False)
and not os.environ.get('ADMIN_PASSWORD', '').strip()
)
error = None
if request.method == 'POST':
_check_csrf()
if not local_auth:
error = gettext('Local password login is disabled. Sign in with your identity provider.')
return render_template('login.html', error=error, next=request.args.get('next', ''),
csrf_token=_get_csrf_token(), temp_password_hint=False,
local_auth_enabled=False,
oidc_enabled=settings.get('oidc_enabled', False),
oidc_display_name=settings.get('oidc_display_name', 'OIDC'))
password = request.form.get('password', '')
pw_hash = settings.get('password_hash', '')
admin_pw = os.environ.get('ADMIN_PASSWORD', '').strip()
if admin_pw:
ok = secrets.compare_digest(password, admin_pw)
else:
ok = bool(pw_hash) and _check_password(password, pw_hash)
if ok:
remember = request.form.get('remember') == 'on'
if settings.get('otp_enabled') and not settings.get('otp_secret') and not admin_pw:
logger.error("Login refused for the admin from %s - two-factor is enabled but "
"its secret could not be decrypted. Restore %s or "
"OTP_ENCRYPTION_KEY, or set ADMIN_PASSWORD to get back in.",
request.remote_addr, env.OTP_KEY_PATH)
session.clear()
return render_template(
'login.html',
error=gettext(
'Two-factor authentication is switched on for this account, but its '
'secret cannot be read, so the code cannot be checked. Signing in with '
'only a password is refused. Restore the secret encryption key, or set '
'ADMIN_PASSWORD to recover access. See the server log for details.'),
next=request.form.get('next', ''),
csrf_token=_get_csrf_token(), temp_password_hint=False,
local_auth_enabled=local_auth,
oidc_enabled=settings.get('oidc_enabled', False),
oidc_display_name=settings.get('oidc_display_name', 'OIDC'))
if settings.get('otp_enabled') and settings.get('otp_secret') and not admin_pw:
session.clear()
session['otp_pending'] = True
session['otp_remember'] = bool(remember)
session['otp_next'] = request.form.get('next') or ''
session['otp_must_change'] = settings.get('must_change_password', False)
session['otp_setup_complete'] = settings.get('setup_complete', False)
session['otp_epoch'] = int(settings.get('session_epoch') or 0)
_auth.start_otp_attempt()
logger.info(f"OTP step required for login from {request.remote_addr}")
return redirect(url_for('login_otp'))
_start_session(remember)
logger.info(f"Successful login from {request.remote_addr}")
if settings.get('must_change_password', False) and not admin_pw:
if not settings.get('setup_complete', False):
return redirect(url_for('setup'))
else:
return redirect(url_for('force_change_password'))
return redirect(_safe_next(request.form.get('next')))
else:
error = gettext('Incorrect password.')
logger.warning(f"Failed login attempt from {request.remote_addr}")
next_url = request.args.get('next', '')
if (request.method == 'GET'
and settings.get('oidc_enabled')
and settings.get('oidc_auto_login')
and request.args.get('auto') != '0'
and not session.get('oidc_auto_tried')):
return redirect(url_for('oidc_login', silent='1'))
return render_template('login.html', error=error, next=next_url,
csrf_token=_get_csrf_token(),
temp_password_hint=temp_password_hint,
local_auth_enabled=local_auth,
oidc_enabled=settings.get('oidc_enabled', False),
oidc_display_name=settings.get('oidc_display_name', 'OIDC'))
@app.route('/setup', methods=['GET', 'POST'])
@limiter.limit("5 per minute", methods=["POST"])
@_setup_otp_failure_limit
def setup():
if not _auth_required():
return redirect(url_for('index'))
unreadable = _settings.settings_unreadable()
if unreadable:
logger.error("Refusing to serve the setup page - %s", unreadable)
return (gettext('Traefik Manager cannot read its configuration file, so it cannot tell '
'whether this is a new install. Setup is disabled until the file is '
'fixed or restored from a backup. See the container log for the path '
'and the reason.'), 503,
{'Content-Type': 'text/plain; charset=utf-8'})
current = load_settings()
reset_mode = bool(current.get('setup_password_reset', False))
otp_required = bool(current.get('otp_enabled') and current.get('otp_secret'))
if not reset_mode:
if current.get('setup_complete', False):
if current.get('must_change_password', False):
return redirect(url_for('force_change_password'))
return redirect(url_for('index'))
if _has_password_set() and not session.get('authenticated'):
return redirect(url_for('login'))
if reset_mode and request.method == 'POST':
_check_csrf()
refusal = shown = None
if os.environ.get('ADMIN_PASSWORD', '').strip():
refusal = 'ADMIN_PASSWORD is set, so a password saved here would never be used. Change that variable and restart instead.'
shown = gettext('ADMIN_PASSWORD is set, so a password saved here would never be used. Change that variable and restart instead.')
elif not _auth_enabled():
refusal = 'Local password login is turned off. Sign in with your identity provider.'
shown = gettext('Local password login is turned off. Sign in with your identity provider.')
if refusal:
update_settings(setup_password_reset=False)
logger.warning(f"Password reset refused from {request.remote_addr}: {refusal}")
flash(shown, 'error')
return redirect(url_for('login'))
new_pw = request.form.get('password', '')
confirm = request.form.get('confirm', '')
err = _password_error(new_pw)
if not err and new_pw != confirm:
err = gettext('Passwords do not match.')
if not err and otp_required:
code = request.form.get('code', '').strip()
try:
import pyotp
code_ok = bool(code) and pyotp.TOTP(current['otp_secret']).verify(code, valid_window=1)
except Exception:
logger.exception("OTP verify error during a password reset")
code_ok = False
if not code_ok:
err = gettext('Enter the current code from your authenticator app.')
logger.warning(f"Password reset with a wrong two-factor code from {request.remote_addr}")
if err:
return render_template('login.html', setup_mode=True, reset_mode=True,
error=err, csrf_token=_get_csrf_token(),
defaults={'domains': current['domains'],
'cert_resolver': current['cert_resolver'],
'traefik_api_url': current['traefik_api_url']},
temp_password_mode=False, otp_required=otp_required,
detected_self_domain='', detected_self_svc='',
detected_self_entry_point='')
_settings.bump_session_epoch(password_hash=_hash_password(new_pw),
must_change_password=False, setup_password_reset=False)
_start_session(False, notify=False)
logger.warning(f"Password reset completed from {request.remote_addr}")
return redirect(url_for('index'))
temp_password_mode = current.get('must_change_password', False) and bool(current.get('password_hash', ''))
defaults = {
'domains': current['domains'],
'cert_resolver': current['cert_resolver'],
'traefik_api_url': current['traefik_api_url'],
}
error = None
if request.method == 'POST':
_check_csrf()
domains_raw = request.form.get('domains', '').strip()
cert_resolver = request.form.get('cert_resolver', '').strip()
traefik_api_url = request.form.get('traefik_api_url', '').strip()
traefik_api_user = request.form.get('traefik_api_user', '').strip()
traefik_api_password = request.form.get('traefik_api_password', '')
visible_tabs_raw = request.form.get('visible_tabs', '{}')
pw = request.form.get('password', '')
confirm = request.form.get('confirm', '')
self_route_domain = request.form.get('self_route_domain', '').strip()
self_route_svc = request.form.get('self_route_service', '').strip() or 'http://traefik-manager:5000'
self_route_ep = request.form.get('self_route_entry_point', '').strip() or _best_entrypoint()
notify_kind = request.form.get('notify_kind', '').strip().lower()
notify_fields = {f: request.form.get('notify_' + f, '').strip()
for f in ('url', 'token', 'token2', 'username', 'password')}
notify_wanted = any(notify_fields.values())
notify_missing = [f for f in _notify_providers.required_fields(notify_kind)
if not notify_fields[f]]
domains = _cfg.parse_domains(domains_raw)
pw_error = None if temp_password_mode else _password_error(pw)
if not domains:
error = gettext('Enter at least one domain.')
elif not traefik_api_url:
error = gettext('Enter the Traefik API URL.')
elif not _safe_api_url(traefik_api_url):
error = gettext('Traefik API URL must start with http:// or https://')
elif pw_error:
error = pw_error
elif not temp_password_mode and pw != confirm:
error = gettext('Passwords do not match.')
elif notify_wanted and notify_kind not in _settings.CHANNEL_KINDS:
error = gettext('Choose a notification destination.')
elif notify_wanted and notify_missing:
error = gettext('Complete every notification field, or clear them to skip notifications.')
else:
import json as _json
try:
vt_raw = _json.loads(visible_tabs_raw)
visible_tabs = {t: bool(vt_raw.get(t, False)) for t in OPTIONAL_TABS}
except Exception:
visible_tabs = {t: False for t in OPTIONAL_TABS}
pw_hash = current['password_hash'] if temp_password_mode else _hash_password(pw)
resolver = cert_resolver if cert_resolver.lower() not in ('none', '') else ''
sr = {'domain': '', 'service_url': ''}
if self_route_domain:
sr = {'domain': self_route_domain, 'service_url': self_route_svc, 'entry_point': self_route_ep}
_write_self_route(self_route_domain, self_route_svc, resolver, entry_point=self_route_ep)
cs_url = request.form.get('crowdsec_lapi_url', '').strip()
git_repo = request.form.get('git_backup_repo', '').strip()
extra = {}
if cs_url:
extra.update(
crowdsec_lapi_url=cs_url,
crowdsec_api_key=request.form.get('crowdsec_api_key', '').strip(),
crowdsec_machine_id=request.form.get('crowdsec_machine_id', '').strip(),
crowdsec_machine_password=request.form.get('crowdsec_machine_password', '').strip(),
)
if git_repo:
extra.update(
git_backup_enabled=True,
git_backup_repo=git_repo,
git_backup_branch=request.form.get('git_backup_branch', '').strip() or 'main',
git_backup_username=request.form.get('git_backup_username', '').strip(),
git_backup_token=request.form.get('git_backup_token', '').strip(),
git_backup_auto_push=request.form.get('git_backup_auto_push', '') == 'on',
)
if notify_wanted:
channel = _blank_channel()
channel.update(kind=notify_kind, name=notify_kind.title(), **notify_fields)
extra['notification_channels'] = list(current.get('notification_channels') or []) + [channel]
if request.form.get('geoip_enabled', '') == 'on':
extra['geoip_enabled'] = True
theme = request.form.get('default_theme', '').strip().lower()
if theme in ('dark', 'light', 'system'):
extra['default_theme'] = theme
if 'default_language' in request.form:
extra['default_language'] = _i18n.normalize(request.form.get('default_language', '')) or ''
save_settings(
domains=domains,
cert_resolver=resolver,
traefik_api_url=traefik_api_url,
traefik_api_user=traefik_api_user,
traefik_api_password=traefik_api_password,
auth_enabled=True,
password_hash=pw_hash,
visible_tabs=visible_tabs,
must_change_password=current.get('must_change_password', False),
setup_complete=True,
self_route=sr,
**extra,
)
logger.info(f"Setup wizard completed from {request.remote_addr}")
if temp_password_mode:
return redirect(url_for('force_change_password'))
_start_session(True, notify=False)
return redirect(url_for('index'))
detected_domain, detected_svc = _detect_setup_self_route()
detected_entry_point = load_settings().get('self_route', {}).get('entry_point', '') or _best_entrypoint()
return render_template('login.html', setup_mode=True, error=error,
reset_mode=reset_mode, otp_required=otp_required,
defaults=defaults, csrf_token=_get_csrf_token(),
temp_password_mode=temp_password_mode,
detected_self_domain=detected_domain,
detected_self_svc=detected_svc,
detected_self_entry_point=detected_entry_point,
detected_tabs=_detect_provider_tabs())
def _setup_open() -> bool:
return not load_settings().get('setup_complete', False)
def _setup_probe_allowed() -> bool:
_auth._check_inactivity()
if not _auth_required() or not _has_password_set():
return True
return bool(session.get('authenticated'))
@app.route('/setup/test-crowdsec', methods=['POST'])
@limiter.limit("10 per minute")
def setup_test_crowdsec():
if not _setup_open() or not _setup_probe_allowed():
abort(404)
_check_csrf()
data = request.get_json(silent=True) or {}
url = str(data.get('url', '')).strip()
key = str(data.get('key', '')).strip()
if not url or not url.startswith(('http://', 'https://')):
return jsonify({'ok': False, 'error': gettext('Enter an http:// or https:// URL')}), 400
if not _ssrf_ok(url):
return jsonify({'ok': False, 'error': gettext('Target address not allowed')}), 400
try:
resp = requests.get(f"{url.rstrip('/')}/v1/decisions",
headers={'X-Api-Key': key, 'Accept': 'application/json'},
timeout=5, allow_redirects=False)
if resp.status_code in (401, 403):
return jsonify({'ok': False, 'error': gettext('Reached the LAPI, but the key was refused')})
resp.raise_for_status()
return jsonify({'ok': True})
except Exception as e:
return jsonify({'ok': False, 'error': str(e)[:120]})
@app.route('/setup/test-git', methods=['POST'])
@limiter.limit("10 per minute")
def setup_test_git():
if not _setup_open() or not _setup_probe_allowed():
abort(404)
_check_csrf()
data = request.get_json(silent=True) or {}
repo_url = str(data.get('repo_url', '')).strip()
token = str(data.get('token', '')).strip()
if not repo_url:
return jsonify({'ok': False, 'error': gettext('No repository URL')}), 400
if not _valid_git_url(repo_url):
return jsonify({'ok': False, 'error': gettext('Unsupported URL - use https://, http://, ssh:// or git://')}), 400
if not _ssrf_ok(repo_url):
return jsonify({'ok': False, 'error': gettext('Target address not allowed')}), 400
creds = {'username': str(data.get('username', '')).strip(), 'token': token} if token else None
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
_, err, rc = _git_run(['ls-remote', '--quiet', '--', repo_url], cwd=tmpdir, credentials=creds,
extra_config=['http.followRedirects=false'])
if rc == 0:
return jsonify({'ok': True})
safe = err.replace(token, '***') if token else err
return jsonify({'ok': False, 'error': safe[:160] if safe else gettext('Could not reach repository')})
@app.route('/logout', methods=['POST'])
@csrf_protect
def logout():
session.clear()
session['oidc_auto_tried'] = True
logger.info(f"User logged out from {request.remote_addr}")
return redirect(url_for('login'))
@app.route('/force-change-password', methods=['GET', 'POST'])
@login_required
def force_change_password():
settings = load_settings()
if not settings.get('must_change_password', False):
return redirect(url_for('index'))
error = None
if request.method == 'POST':
_check_csrf()
new_pw = request.form.get('new_password', '')
confirm = request.form.get('confirm_password', '')
error = _password_error(new_pw)
if not error and new_pw != confirm:
error = gettext('Passwords do not match.')
if not error:
_settings.bump_session_epoch(
password_hash=_hash_password(new_pw),
must_change_password=False,
setup_password_reset=False,
setup_complete=True,
)
_auth._stamp_session()
logger.info(f"Forced password change completed from {request.remote_addr}")
return redirect(url_for('index'))
return render_template('login.html', force_change_mode=True, error=error,
csrf_token=_get_csrf_token())
@app.cli.command('reset-password')
@click.option('--disable-otp', is_flag=True, default=False,
help='Also disable two-factor authentication (use if TOTP app is lost).')
@click.option('--prompt', 'prompt_pw', is_flag=True, default=False,
help='Ask for the new password twice instead of generating a temporary one.')
@click.option('--stdin', 'from_stdin', is_flag=True, default=False,
help='Read the new password from standard input.')
@click.option('--password', 'password_opt', default=None,
help='Set the new password directly. Prefer --stdin for scripts.')
def reset_password_cli(disable_otp, prompt_pw, from_stdin, password_opt):
given = [name for name, used in (('--prompt', prompt_pw),
('--stdin', from_stdin),
('--password', password_opt is not None)) if used]
if len(given) > 1:
raise click.ClickException('Use only one of %s.' % ', '.join(given))
explicit = bool(given)
if prompt_pw:
password = click.prompt('New password', hide_input=True, confirmation_prompt=True,
default='', show_default=False)
elif from_stdin:
try:
raw = click.get_binary_stream('stdin').readline().decode('utf-8-sig')
except UnicodeDecodeError:
raise click.ClickException('The password on standard input is not valid UTF-8.')
password = raw.split('\n', 1)[0].rstrip('\r')
elif password_opt is not None:
click.echo('Warning: --password is visible in ps output and shell history. '
'Use --stdin instead for scripts.', err=True)
password = password_opt
else:
password = secrets.token_urlsafe(16)
if explicit:
if not password:
raise click.ClickException('Password must not be empty.')
pw_error = _password_error(password)
if pw_error:
raise click.ClickException(pw_error)
if os.environ.get('ADMIN_PASSWORD', '').strip():
raise click.ClickException(
'ADMIN_PASSWORD is set, and it overrides the stored password. '
'Change or unset that variable instead, or this password will not work.')
settings = load_settings()
_settings.bump_session_epoch(
password_hash=_hash_password(password),
must_change_password=False if explicit else True,
setup_password_reset=False if explicit else True,
setup_complete=settings.get('setup_complete', True),
otp_secret='' if disable_otp else None,
otp_enabled=False if disable_otp else None,
)
print("=" * 60)
print("TRAEFIK MANAGER - PASSWORD RESET")
if explicit:
print("New password set. It is not shown here.")
else:
print(f"New temporary password: {password}")
if disable_otp:
print("Two-factor authentication has been DISABLED.")
if not explicit:
print("You will be required to change it on next login.")
print("=" * 60)
@app.route('/api/auth/change-password', methods=['POST'])
@limiter.limit("10 per minute")
@csrf_protect
@login_required
def api_change_password():
data = request.get_json()
current_pw = (data or {}).get('current_password', '')
new_pw = (data or {}).get('new_password', '')
confirm_pw = (data or {}).get('confirm_password', '')
pw_error = _password_error(new_pw, new=True)
if pw_error:
return jsonify({'error': pw_error}), 400
if new_pw != confirm_pw:
return jsonify({'error': gettext('Passwords do not match.')}), 400
settings = load_settings()
pw_hash = settings.get('password_hash', '')
admin_pw = os.environ.get('ADMIN_PASSWORD', '').strip()
if admin_pw:
ok = secrets.compare_digest(current_pw, admin_pw)
else:
ok = bool(pw_hash) and _check_password(current_pw, pw_hash)
if not ok:
logger.warning(f"Failed password change attempt from {request.remote_addr}")
return jsonify({'error': gettext('Current password is incorrect.')}), 403
_settings.bump_session_epoch(password_hash=_hash_password(new_pw),
must_change_password=False, setup_password_reset=False)
if session.get('authenticated'):
_auth._stamp_session()
logger.info(f"Password changed successfully from {request.remote_addr}")
return jsonify({'success': True})
@app.route('/api/auth/toggle', methods=['POST'])
@csrf_protect
@login_required
def api_auth_toggle():
data = request.get_json()
enabled = bool((data or {}).get('auth_enabled', True))
settings = load_settings()
save_settings(
domains=settings['domains'],
cert_resolver=settings['cert_resolver'],
traefik_api_url=settings['traefik_api_url'],
auth_enabled=enabled,
password_hash=settings['password_hash'],
visible_tabs=settings['visible_tabs'],
)
logger.info(f"auth_enabled set to {enabled} by {request.remote_addr}")
reauth = _auth_required() and not session.get('authenticated')
return jsonify({'success': True, 'auth_enabled': enabled, 'reauth_required': reauth})
@app.route('/api/auth/external-ack', methods=['POST'])
@csrf_protect
@login_required
def api_auth_external_ack():
data = request.get_json(silent=True) or {}
ack = bool(data.get('auth_external_ack'))
if ack and _auth_required():
return jsonify({'error': gettext('Built-in authentication or OIDC is active, so there is nothing to acknowledge')}), 400
update_settings(auth_external_ack=ack)
logger.warning(f"auth_external_ack set to {ack} by {request.remote_addr} - "
f"the operator asserts this instance is protected by an external provider"
if ack else f"auth_external_ack cleared by {request.remote_addr}")
return jsonify({'success': True, 'auth_external_ack': ack})
@app.route('/login/otp', methods=['GET', 'POST'])
@limiter.limit("5 per minute", methods=["POST"])
@_otp_failure_limit
def login_otp():
if not session.get('otp_pending'):
return redirect(url_for('login'))
if _auth.otp_attempt_expired():
session.clear()
flash(gettext('Your sign-in expired. Enter your password again.'), 'error')
return redirect(url_for('login'))
error = None
if request.method == 'POST':
_check_csrf()
import pyotp
code = request.form.get('code', '').strip()
settings = load_settings()
secret = settings.get('otp_secret', '')
try:
otp_valid = secret and pyotp.TOTP(secret).verify(code, valid_window=1)
except Exception:
logger.exception("OTP verify error - secret may be corrupt")
otp_valid = False
if otp_valid and int(session.get('otp_epoch') or 0) != int(settings.get('session_epoch') or 0):
session.clear()
logger.warning(f"OTP sign-in from {request.remote_addr} abandoned, the password changed after the password step")
flash(gettext('Your password was changed while you were signing in. Sign in again.'), 'error')
return redirect(url_for('login'))
if otp_valid:
remember = session.get('otp_remember', True)
must_change = session.get('otp_must_change', False)
setup_complete = session.get('otp_setup_complete', False)
next_url = session.get('otp_next', '') or url_for('index')
_auth.forget_otp_attempt()
_start_session(remember)
logger.info(f"Successful OTP login from {request.remote_addr}")
if must_change:
if not setup_complete:
return redirect(url_for('setup'))
return redirect(url_for('force_change_password'))
return redirect(_safe_next(next_url))
else:
logger.warning(f"Failed OTP attempt from {request.remote_addr}")
if _auth.record_otp_failure() >= _auth.OTP_MAX_ATTEMPTS:
session.clear()
flash(gettext('Too many wrong codes. Enter your password again.'), 'error')
return redirect(url_for('login'))
error = gettext('Invalid code. Please try again.')
return render_template('login.html', otp_mode=True, error=error,
csrf_token=_get_csrf_token())
@app.route('/api/auth/otp/setup', methods=['POST'])
@csrf_protect
@login_required
def api_otp_setup():
import pyotp
secret = pyotp.random_base32()
uri = pyotp.TOTP(secret).provisioning_uri(
name='Traefik Manager',
issuer_name='traefik-manager'
)
session['otp_pending_secret'] = secret
return jsonify({'secret': secret, 'uri': uri})
@app.route('/api/auth/otp/enable', methods=['POST'])
@csrf_protect
@login_required
def api_otp_enable():
import pyotp
code = (request.get_json() or {}).get('code', '').strip()
secret = session.pop('otp_pending_secret', '')
if not secret or not pyotp.TOTP(secret).verify(code, valid_window=1):
return jsonify({'error': gettext('Invalid code - please try again.')}), 400
update_settings(otp_secret=secret,
otp_enabled=True)
logger.info(f"OTP enabled by {request.remote_addr}")
return jsonify({'success': True})
@app.route('/api/auth/otp/disable', methods=['POST'])
@csrf_protect
@login_required
def api_otp_disable():
_settings.bump_session_epoch(otp_secret='', otp_enabled=False)
if session.get('authenticated'):
_auth._stamp_session()
logger.info(f"OTP disabled by {request.remote_addr}")
return jsonify({'success': True})
@app.route('/api/auth/sessions/revoke', methods=['POST'])
@csrf_protect
@login_required
def api_revoke_sessions():
_settings.bump_session_epoch()
if session.get('authenticated'):
_auth._stamp_session()
logger.warning(f"Every other session was signed out from {request.remote_addr}")
add_notification('warning', _i18n.lazy_gettext('Every other session was signed out from %(ip)s',
ip=request.remote_addr), category='security')
return jsonify({'success': True})
@app.route('/api/auth/otp/status')
@login_required
def api_otp_status():
settings = load_settings()
enabled = bool(settings.get('otp_enabled', False))
unusable = bool(enabled and not settings.get('otp_secret', ''))
return jsonify({'otp_enabled': enabled,
'otp_secret_unreadable': unusable})
@app.route('/api/auth/apikey/generate', methods=['POST'])
@limiter.limit("5 per hour")
@csrf_protect
@login_required
def api_apikey_generate():
data = request.get_json(silent=True) or {}
device_name = str(data.get('device_name', '')).strip()[:50]
if not device_name:
return jsonify({'ok': False, 'error': gettext('device_name is required')}), 400
settings = load_settings()
api_keys = settings.get('api_keys', [])
if len(api_keys) >= 10:
return jsonify({'ok': False, 'error': gettext('Maximum of 10 API keys reached')}), 400
key = secrets.token_urlsafe(32)
preview = key[:8] + '...' + key[-4:]
api_keys.append({
'name': device_name,
'hash': _hash_api_key(key),
'preview': preview,
'created_at': datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M'),
})
update_settings(otp_secret=settings['otp_secret'],
otp_enabled=settings['otp_enabled'],
api_keys=api_keys)
logger.info(f"API key '{device_name}' generated by {request.remote_addr}")
return jsonify({'ok': True, 'key': key})
@app.route('/api/auth/apikey/revoke', methods=['POST'])
@csrf_protect
@login_required
def api_apikey_revoke():
data = request.get_json(silent=True) or {}
preview = str(data.get('preview', '')).strip()
if not preview:
return jsonify({'ok': False, 'error': gettext('preview is required')}), 400
settings = load_settings()
api_keys = [k for k in settings.get('api_keys', []) if k.get('preview') != preview]
update_settings(otp_secret=settings['otp_secret'],
otp_enabled=settings['otp_enabled'],
api_keys=api_keys)
logger.info(f"API key revoked by {request.remote_addr}")
return jsonify({'ok': True})
@app.route('/api/auth/apikey/status')
@login_required
def api_apikey_status():
settings = load_settings()
api_keys = settings.get('api_keys', [])
return jsonify({
'enabled': len(api_keys) > 0,
'keys': [{'name': k['name'], 'preview': k['preview'], 'created_at': k.get('created_at', '')} for k in api_keys],
'count': len(api_keys),
})
if not os.environ.get('REQUESTS_CA_BUNDLE'):
_SYSTEM_CA_BUNDLE = '/etc/ssl/certs/ca-certificates.crt'
if os.path.exists(_SYSTEM_CA_BUNDLE):
os.environ['REQUESTS_CA_BUNDLE'] = _SYSTEM_CA_BUNDLE
@app.route('/api/traefik/overview')
@login_required
def api_overview():
return jsonify(traefik_api_get('/api/overview') or {})
def _traefik_proto_payload(kind):
fetched = {}
complete = []
for proto in ('http', 'tcp', 'udp'):
fetched[proto] = traefik_api_get_all(f'/api/{proto}/{kind}', complete)
out = {p: (v or []) for p, v in fetched.items()}
out['reachable'] = any(v is not None for v in fetched.values())
out['complete'] = all(v is not None for v in fetched.values()) and not complete
return out
@app.route('/api/traefik/routers')
@login_required
def api_routers():
return jsonify(_traefik_proto_payload('routers'))
@app.route('/api/services/<path:name>/ownership', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_service_ownership(name):
data = request.get_json(silent=True) or {}
adopt = bool(data.get('adopt'))
bare = str(name).split('@')[0]
agent_id, agent, err = _svc_agent_ctx()
if err:
return err
if agent:
pairs = [(fname, cfg) for fname, cfg in _agent_load_configs(agent).items()]
else:
pairs = [(os.path.basename(p), load_config(p)) for p in env.CONFIG_PATHS]
svc_def, cfg_file = None, ''
for fname, cfg in pairs:
found = ((cfg.get('http') or {}).get('services') or {}).get(bare)
if isinstance(found, dict):
svc_def, cfg_file = found, fname
break
if svc_def is None:
return jsonify({'ok': False, 'error': gettext('Service not found')}), 404
if adopt and not _svc_own.composite_type(svc_def):
return jsonify({'ok': False,
'error': gettext('Only weighted, mirroring, failover and highestRandomWeight services can be managed here')}), 400
settings = load_settings()
ledger = dict(settings.get('managed_middlewares') or {})
key = _svc_own.ledger_key(bare, agent_id)
if adopt:
ledger[key] = _svc_own.ledger_entry(svc_def, cfg_file)
elif key in ledger:
del ledger[key]
else:
return jsonify({'ok': True, 'owned': False})
_save_edit_dicts(managed_middlewares=ledger)
logger.info(f"Service {bare!r} {'adopted' if adopt else 'released'} by {request.remote_addr}")
return jsonify({'ok': True, 'owned': adopt})
def _disabled_router_name(key: str) -> str:
return key.split('::', 1)[1] if '::' in key else key
def _service_routers_using(configs, name: str) -> list:
out = []
for cfg in configs:
for section in ('http', 'tcp', 'udp'):
for rname, rdata in ((cfg.get(section) or {}).get('routers') or {}).items():
if isinstance(rdata, dict) and str(rdata.get('service') or '').split('@')[0] == name:
out.append(rname)
return out
def _stream_service_proto(name: str) -> str:
bare = str(name or '').split('@')[0]
if not bare:
return ''
for path in env.CONFIG_PATHS:
config = load_config(path)
for section, label in (('tcp', 'TCP'), ('udp', 'UDP')):
if bare in ((config.get(section) or {}).get('services') or {}):
return label
return ''
def _remove_router(config, sec, name, ledger, agent_id=''):
s = config.get(sec) or {}
routers = s.get('routers') or {}
if name not in routers:
return False
svc = str((routers[name] or {}).get('service') or '').strip()
del routers[name]
if svc and 'services' in s and svc in s['services'] and not _service_shared(config, svc, name):
del s['services'][svc]
_drop_owned_transport(config, svc, ledger, agent_id)
if sec == 'http':
for gone in _composite.drop_orphan_children(s['services'], name, set()):
ledger.pop(_svc_ledger_key(gone, agent_id), None)
ledger.pop(_svc_ledger_key(svc, agent_id), None)
return True
def _strip_service_child(config, parent, child):
sdef = ((config.get('http') or {}).get('services') or {}).get(parent)
if not isinstance(sdef, dict):
return ''
hit = False
for kind in ('weighted', 'mirroring', 'failover', 'highestRandomWeight'):
block = sdef.get(kind)
if not isinstance(block, dict):
continue
for key in ('services', 'mirrors'):
lst = block.get(key)
if isinstance(lst, list):
kept = [c for c in lst if not (isinstance(c, dict)
and str(c.get('name') or '').split('@')[0] == child)]
if len(kept) != len(lst):
hit = True
block[key] = kept
for key in ('service', 'fallback'):
if str(block.get(key) or '').split('@')[0] == child:
hit = True
block.pop(key, None)
if not hit:
return ''
return 'emptied' if not _svc_own.child_names(sdef) else 'stripped'
def _service_referenced_by(configs, name: str) -> list:
out = []
for cfg in configs:
for sname, sdef in ((cfg.get('http') or {}).get('services') or {}).items():
if sname == name:
continue
if name in _svc_own.child_names(sdef):
out.append(sname)
return out
def _service_home_path(name: str) -> str:
bare = str(name or '').split('@')[0]
if not bare:
return ''
for path in env.CONFIG_PATHS:
if bare in ((load_config(path).get('http') or {}).get('services') or {}):
return path
return ''
def _children_claimed_by_another(ledger, parent: str, agent_id: str = ''):
prefix = f'{parent}-backend-'
for key, value in (ledger or {}).items():
if not (isinstance(key, str) and isinstance(value, dict)):
continue
bare = key.split('svc::', 1)[-1]
if not bare.startswith(prefix) or not value.get('child'):
continue
owner = value.get('parent')
if owner and owner != parent:
return (bare, owner)
return None
def _children_still_in_use(configs, parent: str, keep) -> list:
prefix = f'{parent}-backend-'
keep = set(keep or ())
blocked = []
for cfg in configs:
for child in ((cfg.get('http') or {}).get('services') or {}):
if not (isinstance(child, str) and child.startswith(prefix)) or child in keep:
continue
users = _service_routers_using(configs, child)
owners = [o for o in _service_referenced_by(configs, child) if o != parent]
if users or owners:
blocked.append((child, sorted(set(users + owners))))
return blocked
def _in_use_error(blocked):
child, users = blocked[0]
return jsonify({'ok': False,
'error': gettext('%(child)s is still used by %(users)s', child=child, users=_name_list(users))}), 409
def _error_param(exc):
args = getattr(exc, 'args', ())
if len(args) == 1 and isinstance(args[0], _i18n.Message):
return args[0]
return str(exc)
def _name_list(names, limit: int = 5) -> str:
names = list(names or [])
shown = ', '.join(str(n) for n in names[:limit])
return f'{shown} (+{len(names) - limit})' if len(names) > limit else shown
def _agent_service_home(agent_configs: dict, name: str) -> str:
bare = str(name or '').split('@')[0]
if not bare:
return ''
for fname, cfg in (agent_configs or {}).items():
if bare in (((cfg or {}).get('http') or {}).get('services') or {}):
return fname
return ''
def _svc_agent_ctx():
agent_id = (request.args.get('agent_id')
or (request.get_json(silent=True) or {}).get('agent_id') or '').strip()
if not agent_id:
return '', None, None
agent = _agent_by_id(agent_id)
if not agent:
return agent_id, None, (jsonify({'ok': False, 'error': gettext('Agent not found')}), 404)
return agent_id, agent, None
@app.route('/api/services', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_service_save():
data = request.get_json(silent=True) or {}
name = str(data.get('name') or '').strip()
kind = str(data.get('type') or '').strip()
original = str(data.get('originalName') or '').strip()
cfg_raw = str(data.get('configFile') or '').strip()
children = data.get('children') or []
name_err = _naming.name_error(name)
if name_err:
return jsonify({'ok': False, 'error': name_err}), 400
if kind not in _composite.TYPES + ('loadBalancer',):
return jsonify({'ok': False,
'error': gettext('Choose load balancer, weighted, mirroring or failover')}), 400
if kind == 'failover' and len(_composite.normalise_children(children)) > 2:
return jsonify({'ok': False,
'error': gettext('Failover takes two backends: the one that serves and the one that takes over')}), 400
hc_sent = 'healthCheck' in data and kind == 'loadBalancer'
lb_extra = {'healthCheck': _healthcheck_block(data.get('healthCheck'))} if hc_sent else None
block, owned, _names = _composite.build(name, kind, children, lb_extra=lb_extra)
if not block:
return jsonify({'ok': False, 'error': gettext('Add at least one backend')}), 400
clash = _stream_service_proto(name) or (_stream_service_proto(original) if original else '')
if clash:
return jsonify({'ok': False,
'error': gettext('That name belongs to a %(clash)s service, and only HTTP services can be edited here', clash=clash)}), 400
agent_id, agent, err = _svc_agent_ctx()
if err:
return err
if agent:
agent_configs = _agent_load_configs(agent)
cfg_filename = _agent_service_home(agent_configs, original or name) \
or (cfg_raw if cfg_raw in agent_configs else next(iter(agent_configs), 'dynamic.yml'))
config = agent_configs.get(cfg_filename) or {}
target_path = None
else:
target_path = _service_home_path(original or name) \
or (_resolve_config_path(cfg_raw) if cfg_raw else env.CONFIG_PATH)
if not target_path:
return jsonify({'ok': False, 'error': gettext("Cannot write to '%(config_file)s'", config_file=cfg_raw)}), 400
cfg_filename = os.path.basename(target_path)
config = load_config(target_path)
section = config.setdefault('http', {}).setdefault('services', {})
settings = load_settings()
ledger = dict(settings.get('managed_middlewares') or {})
existing = section.get(name)
if isinstance(existing, dict) and name != original \
and not _svc_own.is_owned(name, existing, ledger, agent_id) \
and not (kind == 'loadBalancer' and 'loadBalancer' in existing):
return jsonify({'ok': False, 'error': gettext("A service named '%(name)s' already exists", name=name)}), 409
if original and original != name:
_orig_def = section.get(original)
if not _svc_own.is_owned(original, _orig_def, ledger, agent_id) \
and not (isinstance(_orig_def, dict) and 'loadBalancer' in _orig_def):
return jsonify({'ok': False, 'error': gettext('That service is not managed here')}), 403
section.pop(original, None)
_retarget_service(config, original, name)
for gone in _composite.drop_orphan_children(section, original, set()):
ledger.pop(_svc_ledger_key(gone, agent_id), None)
ledger.pop(_svc_ledger_key(original, agent_id), None)
blocked = _children_still_in_use([config], name, owned)
if blocked:
return _in_use_error(blocked)
claimed = _children_claimed_by_another(ledger, name, agent_id)
if claimed:
return jsonify({'ok': False,
'error': gettext('%(claimed)s already belongs to %(claimed2)s. Pick a different name', claimed=claimed[0], claimed2=claimed[1])}), 409
_loop = _composite.find_cycle(section, name, children)
if _loop:
return jsonify({'ok': False,
'error': (gettext('%(name)s cannot use itself as a backend', name=name) if _loop == name
else gettext('%(loop)s already routes back to %(name)s, which would make a cycle Traefik cannot load', loop=_loop, name=name))}), 400
if hc_sent:
_prev_def = section.get(original or name)
_prev_hc = (_prev_def.get('loadBalancer') or {}).get('healthCheck') \
if isinstance(_prev_def, dict) and isinstance(_prev_def.get('loadBalancer'), dict) else None
_new_hc = (block.get('loadBalancer') or {}).get('healthCheck')
if isinstance(_prev_hc, dict) and isinstance(_new_hc, dict):
for _k, _v in _prev_hc.items():
if _k not in _HC_KNOWN:
_new_hc.setdefault(_k, _v)
_composite.merge_into(section, name, block, owned,
extra_authored=('healthCheck',) if hc_sent else ())
for gone in _composite.drop_orphan_children(section, name, set(owned)):
ledger.pop(_svc_ledger_key(gone, agent_id), None)
if kind in _composite.TYPES:
ledger.update(_composite.ledger_entries(name, block, owned, cfg_filename, agent_id))
else:
ledger.pop(_svc_ledger_key(name, agent_id), None)
if agent:
_agent_write_config(agent, cfg_filename, config)
else:
create_backup(target_path)
save_config(_strip_empty_sections(config), target_path)
if original and original != name:
_cascade_across_configs(agent, lambda c: _retarget_service(c, original, name),
already=cfg_filename if agent else target_path)
_save_edit_dicts(managed_middlewares=ledger)
logger.info(f"Service {name!r} saved by {request.remote_addr}")
add_notification('success', _i18n.lazy_gettext('Service %(name)s saved', name=name), category='config')
if agent:
threading.Thread(target=lambda: _git_push_agent_if_enabled(agent, 'service save'),
daemon=True).start()
else:
threading.Thread(target=lambda: _git_push_if_enabled('service save'), daemon=True).start()
return jsonify({'ok': True, 'name': name})
@app.route('/api/services/<path:name>', methods=['DELETE'])
@csrf_protect
@login_required
@_config_edit
def api_service_delete(name):
bare = str(name).split('@')[0]
agent_id, agent, err = _svc_agent_ctx()
if err:
return err
agent_configs = _agent_load_configs(agent) if agent else {}
configs = list(agent_configs.values()) if agent else [load_config(p) for p in env.CONFIG_PATHS]
force = str(request.args.get('force') or '').strip().lower() in ('1', 'true', 'yes')
used_by = sorted(set(_service_routers_using(configs, bare)))
parents = sorted(set(_service_referenced_by(configs, bare)))
if (used_by or parents) and not force:
if used_by and parents:
error = gettext('%(child)s is still used by %(users)s and is still a backend of %(parents)s',
child=bare, users=_name_list(used_by), parents=_name_list(parents))
elif used_by:
error = gettext('%(child)s is still used by %(users)s', child=bare, users=_name_list(used_by))
else:
error = gettext('%(child)s is still a backend of %(parents)s', child=bare, parents=_name_list(parents))
return jsonify({'ok': False, 'error': error, 'inUseBy': used_by, 'parents': parents}), 409
settings = load_settings()
ledger = dict(settings.get('managed_middlewares') or {})
targets = ([(fname, cfg) for fname, cfg in agent_configs.items()] if agent
else [(path, load_config(path)) for path in env.CONFIG_PATHS])
home = None
for idx, (where, config) in enumerate(targets):
section = (config.get('http') or {}).get('services') or {}
if bare in section:
home = idx
_def = section.get(bare)
if not _svc_own.is_owned(bare, _def, ledger, agent_id) \
and not (isinstance(_def, dict) and 'loadBalancer' in _def):
return jsonify({'ok': False, 'error': gettext('That service is not managed here')}), 403
break
if home is None:
return jsonify({'ok': False, 'error': gettext('Service not found')}), 404
if not force:
child_users = _children_still_in_use(configs, bare, set())
if child_users:
return _in_use_error(child_users)
touched = set()
deleted_routers, deleted_services = [], []
victims, frontier = {bare}, [bare]
while frontier:
victim = frontier.pop()
for idx, (where, config) in enumerate(targets):
for sec in ('http', 'tcp', 'udp'):
for rname in list(((config.get(sec) or {}).get('routers') or {})):
rd = config[sec]['routers'].get(rname)
if isinstance(rd, dict) and str(rd.get('service') or '').split('@')[0] == victim:
if _remove_router(config, sec, rname, ledger, agent_id):
deleted_routers.append(rname)
touched.add(idx)
for pname in list(((config.get('http') or {}).get('services') or {})):
if pname in victims:
continue
res = _strip_service_child(config, pname, victim)
if res:
touched.add(idx)
if res == 'emptied':
victims.add(pname)
frontier.append(pname)
for victim in victims:
for idx, (where, config) in enumerate(targets):
section = (config.get('http') or {}).get('services') or {}
if victim not in section:
continue
section.pop(victim, None)
keep = {c for c, _u in _children_still_in_use([cfg for _w, cfg in targets], victim, set())}
for gone in _composite.drop_orphan_children(section, victim, keep):
ledger.pop(_svc_ledger_key(gone, agent_id), None)
ledger.pop(_svc_ledger_key(victim, agent_id), None)
deleted_services.append(victim)
touched.add(idx)
for idx in sorted(touched):
where, config = targets[idx]
if agent:
_agent_write_config(agent, where, config)
else:
create_backup(where)
save_config(_strip_empty_sections(config), where)
_save_edit_dicts(managed_middlewares=ledger)
logger.info(f"Service {bare!r} deleted by {request.remote_addr}")
add_notification('warning', _i18n.lazy_gettext('Service %(name)s deleted', name=bare), category='config')
if agent:
threading.Thread(target=lambda: _git_push_agent_if_enabled(agent, 'service delete'),
daemon=True).start()
else:
threading.Thread(target=lambda: _git_push_if_enabled('service delete'), daemon=True).start()
return jsonify({'ok': True, 'deleted': {'routers': sorted(set(deleted_routers)),
'services': sorted(set(deleted_services))}})
def _owned_child_services(agent_id: str = '') -> list:
prefix = f'agent_{agent_id}::svc::' if agent_id else 'svc::'
ledger = load_settings().get('managed_middlewares') or {}
return sorted(
key[len(prefix):] for key, value in ledger.items()
if isinstance(key, str) and key.startswith(prefix)
and isinstance(value, dict) and value.get('kind') == _svc_own.LEDGER_KIND
and value.get('child') is True)
def _owned_parent_services(agent_id: str = '') -> list:
prefix = f'agent_{agent_id}::svc::' if agent_id else 'svc::'
ledger = load_settings().get('managed_middlewares') or {}
return sorted(
key[len(prefix):] for key, value in ledger.items()
if isinstance(key, str) and key.startswith(prefix)
and isinstance(value, dict) and value.get('kind') == _svc_own.LEDGER_KIND
and value.get('child') is not True)
def _prune_service_ledger(agent_id: str = ''):
ledger = load_settings().get('managed_middlewares') or {}
if not any(isinstance(k, str) and 'svc::' in k for k in ledger):
return
with _locks.config_edit_lock(f'config:agent:{agent_id}' if agent_id else 'config:local'):
if _get_config_parse_errors():
return
configs = [load_config(p) for p in env.CONFIG_PATHS]
if not any(cfg for cfg in configs):
return
def prune(current):
kept, dropped = _svc_own.prune(current.get('managed_middlewares') or {}, configs, agent_id)
return {'managed_middlewares': kept} if dropped else None
_settings.modify_settings(prune)
@app.route('/api/traefik/services')
@login_required
def api_services():
agent_id, agent, err = _svc_agent_ctx()
if err:
return err
if agent:
resp = _agent_request(agent, 'GET', '/api/traefik/services')
payload = resp.json() if resp.ok else {'http': [], 'tcp': [], 'udp': [],
'reachable': False}
else:
payload = _traefik_proto_payload('services')
_prune_service_ledger()
payload['ownedChildren'] = _owned_child_services(agent_id)
payload['ownedServices'] = _owned_parent_services(agent_id)
return jsonify(payload)
@app.route('/api/traefik/middlewares')
@login_required
def api_middlewares():
http_mws = traefik_api_get_all('/api/http/middlewares')
tcp_mws = traefik_api_get_all('/api/tcp/middlewares')
if http_mws is None and tcp_mws is None:
return jsonify({'error': gettext('Traefik API unreachable')}), 502
return jsonify({
'http': http_mws or [],
'tcp': tcp_mws or [],
})
@app.route('/api/manager/router-names')
@login_required
def api_manager_router_names():
server = str(request.args.get('server', '')).strip()
if server:
agent = _agent_by_id(server)
if not agent:
return jsonify({'error': gettext('Unknown server')}), 404
configs = list(_agent_load_configs(agent).values())
else:
configs = [load_config()]
names = set()
for config in configs:
for proto in ('http', 'tcp', 'udp'):
names.update(((config or {}).get(proto) or {}).get('routers', {}).keys())
return jsonify(list(names))
@app.route('/api/traefik/entrypoints')
@login_required
def api_entrypoints():
eps = traefik_api_get_all('/api/entrypoints')
if eps is None:
return jsonify({'error': gettext('Traefik API unreachable')}), 502
return jsonify(eps)
@app.route('/api/traefik/version')
@login_required
def api_version():
return jsonify(traefik_api_get('/api/version') or {})
CS_PAGE_SIZE = 1000
CS_MAX_PAGES = 200
def _cs_stale_note(seconds: int, why: str) -> str:
if seconds >= 86400:
return _i18n.lazy_ngettext('CrowdSec has not answered for %(num)d day, so these decisions are the last ones '
'read and may be out of date. %(reason)s',
'CrowdSec has not answered for %(num)d days, so these decisions are the last ones '
'read and may be out of date. %(reason)s', seconds // 86400, reason=why)
if seconds >= 3600:
return _i18n.lazy_ngettext('CrowdSec has not answered for %(num)d hour, so these decisions are the last ones '
'read and may be out of date. %(reason)s',
'CrowdSec has not answered for %(num)d hours, so these decisions are the last '
'ones read and may be out of date. %(reason)s', seconds // 3600, reason=why)
return _i18n.lazy_ngettext('CrowdSec has not answered for %(num)d minute, so these decisions are the last ones '
'read and may be out of date. %(reason)s',
'CrowdSec has not answered for %(num)d minutes, so these decisions are the last ones '
'read and may be out of date. %(reason)s', max(1, seconds // 60), reason=why)
def _cs_decisions_gate():
lapi = _cs_lapi_url()
key = _cs_api_key()
if not lapi:
return jsonify({'error': gettext('CrowdSec not configured')}), 503
if not key and not _cs_has_cert():
return jsonify({'error': gettext('No bouncer API key or client certificate. CrowdSec only accepts a bouncer key or a TLS client certificate on /v1/decisions, the machine token is refused there')}), 503
return None
def _cs_active_decisions(force_full: bool = False):
lapi = _cs_lapi_url()
key = _cs_api_key()
all_decisions = None
stale_note = ''
if _crowd._cs_stream_cache.get('streamable', True):
try:
all_decisions, _mode = _crowd.cs_decisions_stream(force_full=force_full)
if str(_mode).startswith('stale:'):
_, _age, _why = str(_mode).split(':', 2)
stale_note = _cs_stale_note(int(getattr(_mode, 'age', _age)), getattr(_mode, 'reason', _why))
except CrowdSecUnavailable as e:
if 'HTTP 404' in str(e) or 'HTTP 405' in str(e):
logger.info("CrowdSec LAPI has no /v1/decisions/stream, falling back to the paged walk")
_crowd._cs_stream_cache['streamable'] = False
all_decisions = None
else:
raise
if all_decisions is None:
all_decisions = []
cursor = 0
for _page in range(CS_MAX_PAGES):
try:
chunk = _cs_request_strict('GET', f'/v1/decisions?limit={CS_PAGE_SIZE}&id_gt={cursor}',
lapi=lapi, key=key)
except CrowdSecUnavailable:
if all_decisions:
logger.warning(f"CrowdSec decisions walk failed at page {_page + 1}, "
f"returning the {len(all_decisions)} rows already read")
break
raise
if not isinstance(chunk, list) or not chunk:
break
all_decisions.extend(chunk)
ids = [d.get('id') for d in chunk if isinstance(d.get('id'), int)]
if not ids:
break
cursor = max(ids)
if len(chunk) < CS_PAGE_SIZE:
break
now = datetime.now(timezone.utc)
active = []
for d in all_decisions:
until = d.get('until')
if until:
try:
exp = datetime.fromisoformat(until.replace('Z', '+00:00'))
if exp < now:
continue
except Exception:
pass
active.append(d)
return active, stale_note
def _cs_origin_key(d: dict) -> str:
return str(d.get('origin') or '').strip().lower()
def _cs_is_subscribed(d: dict) -> bool:
return _cs_origin_key(d) in ('capi', 'lists')
def _cs_is_own(d: dict) -> bool:
return not _cs_is_subscribed(d)
def _cs_is_byhand(d: dict) -> bool:
return _cs_origin_key(d) in ('cscli', 'manual')
@app.route('/api/crowdsec/decisions')
@login_required
def api_cs_decisions():
gate = _cs_decisions_gate()
if gate:
return gate
force_full = request.args.get('full') in ('1', 'true', 'yes')
try:
active, stale_note = _cs_active_decisions(force_full)
if stale_note:
logger.warning(f"CrowdSec decisions served from a stale cache: {stale_note}")
resp = jsonify(active)
resp.headers['X-CS-Stale'] = stale_note
return resp
return jsonify(active)
except CrowdSecUnavailable as e:
return jsonify({'error': _i18n.shown_error(e)}), 502
except Exception as e:
logger.exception("CrowdSec decisions error")
return jsonify({'error': _i18n.shown_error(e)}), 500
CS_SEARCH_PER_DEFAULT = 20
CS_SEARCH_PER_MAX = 200
@app.route('/api/crowdsec/decisions/search')
@login_required
def api_cs_decisions_search():
gate = _cs_decisions_gate()
if gate:
return gate
try:
active, stale_note = _cs_active_decisions()
except CrowdSecUnavailable as e:
return jsonify({'error': _i18n.shown_error(e)}), 502
except Exception as e:
logger.exception("CrowdSec decisions search error")
return jsonify({'error': _i18n.shown_error(e)}), 500
q = request.args.get('q', '').strip().lower()
origin_f = request.args.get('origin', '').strip()
type_f = request.args.get('type', '').strip().lower()
ip_f = request.args.get('ip', '').strip()
scenario_f = request.args.get('scenario', '').strip()
try:
page = int(request.args.get('page', 1))
except ValueError:
page = 1
page = max(1, page)
try:
per = int(request.args.get('per', CS_SEARCH_PER_DEFAULT))
except ValueError:
per = CS_SEARCH_PER_DEFAULT
per = max(1, min(CS_SEARCH_PER_MAX, per))
def _match(d, use_origin=True, use_type=True, use_ip=True, use_scenario=True, use_q=True):
if use_origin and origin_f:
if origin_f == 'subscribed':
ok = _cs_is_subscribed(d)
elif origin_f == 'own':
ok = _cs_is_own(d)
elif origin_f == 'byhand':
ok = _cs_is_byhand(d)
else:
ok = _cs_origin_key(d) == origin_f.strip().lower()
if not ok:
return False
if use_type and type_f and str(d.get('type') or '').strip().lower() != type_f:
return False
if use_ip and ip_f and d.get('value') != ip_f:
return False
if use_scenario and scenario_f and d.get('scenario') != scenario_f:
return False
if use_q and q:
hay = ' '.join([str(d.get('value') or ''), str(d.get('scenario') or ''), _cs_origin_key(d),
str(d.get('scope') or ''), str(d.get('type') or '')]).lower()
if q not in hay:
return False
return True
rows = [d for d in active if _match(d)]
rows.sort(key=lambda d: (0 if _cs_is_own(d) else 1, -(d.get('id') or 0)))
total = len(rows)
per_pages = -(-total // per) if total else 0
pages = max(1, per_pages)
page = min(page, pages)
start = (page - 1) * per
page_rows = rows[start:start + per]
facet_totals = {
'origin': sum(1 for d in active if _match(d, use_type=False, use_ip=False, use_scenario=False)),
'type': sum(1 for d in active if _match(d, use_origin=False, use_ip=False, use_scenario=False)),
}
resp = jsonify({'rows': page_rows, 'total': total, 'page': page, 'pages': pages,
'per': per, 'facet_totals': facet_totals})
if stale_note:
resp.headers['X-CS-Stale'] = stale_note
return resp
CS_SUMMARY_ROW_CAP = 500
CS_SUMMARY_ALERT_KEYS = ('id', 'uuid', 'scenario', 'scenario_version', 'events_count', 'capacity',
'leakspeed', 'simulated', 'machine_id', 'message', 'start_at', 'stop_at',
'created_at', 'source', 'meta')
@app.route('/api/crowdsec/summary')
@login_required
def api_cs_summary():
lapi = _cs_lapi_url()
if not lapi:
return jsonify({'error': gettext('CrowdSec not configured')}), 503
force_full = request.args.get('full') in ('1', 'true', 'yes')
key = _cs_api_key()
decisions_block = {'ok': True, 'error': '', 'stale': '', 'total': 0, 'own': 0, 'subscribed': 0,
'wide': 0, 'origins': {}, 'types': {}, 'rows': [], 'rows_more': 0}
active = []
if not key and not _cs_has_cert():
decisions_block['ok'] = False
decisions_block['error'] = gettext('No bouncer API key or client certificate. CrowdSec only accepts a bouncer key or a TLS client certificate on /v1/decisions, the machine token is refused there')
else:
try:
active, stale_note = _cs_active_decisions(force_full)
decisions_block['stale'] = _i18n.shown(stale_note)
except CrowdSecUnavailable as e:
decisions_block['ok'] = False
decisions_block['error'] = _i18n.shown_error(e)
except Exception as e:
logger.exception("CrowdSec summary decisions error")
decisions_block['ok'] = False
decisions_block['error'] = _i18n.shown_error(e)
if decisions_block['ok']:
origins = {}
types = {}
own_rows = []
own_count = 0
subscribed_count = 0
wide_count = 0
for d in active:
okey = _cs_origin_key(d)
origins[okey] = origins.get(okey, 0) + 1
tkey = str(d.get('type') or '').strip().lower()
types[tkey] = types.get(tkey, 0) + 1
if _cs_is_subscribed(d):
subscribed_count += 1
else:
own_count += 1
if okey != 'crowdsec':
own_rows.append(d)
if str(d.get('scope') or '') != 'Ip':
wide_count += 1
own_rows.sort(key=lambda d: -(d.get('id') or 0))
decisions_block['total'] = len(active)
decisions_block['own'] = own_count
decisions_block['subscribed'] = subscribed_count
decisions_block['wide'] = wide_count
decisions_block['origins'] = origins
decisions_block['types'] = types
decisions_block['rows'] = own_rows[:CS_SUMMARY_ROW_CAP]
decisions_block['rows_more'] = max(0, len(own_rows) - CS_SUMMARY_ROW_CAP)
alert_limit = cs_alert_limit()
alerts_block = {'ok': True, 'error': '', 'status': 200, 'limit': alert_limit, 'capped': False, 'rows': []}
alert_rows_raw = []
try:
alert_rows_raw, _mode = _crowd.cs_alerts(alert_limit, force_full=force_full)
alerts_block['capped'] = bool(alert_limit and len(alert_rows_raw) >= alert_limit)
except CrowdSecUnavailable as e:
alerts_block['ok'] = False
alerts_block['error'] = _i18n.shown_error(e)
alerts_block['status'] = getattr(e, 'status', 0) or 0
except Exception as e:
logger.exception("CrowdSec summary alerts error")
alerts_block['ok'] = False
alerts_block['error'] = _i18n.shown_error(e)
alerts_block['status'] = 0
if alerts_block['ok']:
handled_values = set()
if decisions_block['ok']:
handled_values = {d.get('value') for d in active if d.get('scope') in ('Ip', 'Range')}
trimmed = []
for a in alert_rows_raw:
row = {k: a[k] for k in CS_SUMMARY_ALERT_KEYS if k in a}
if decisions_block['ok']:
src = a.get('source') or {}
row['handled'] = (src.get('ip') or src.get('value')) in handled_values
trimmed.append(row)
trimmed.sort(key=lambda r: -(r.get('id') or 0))
alerts_block['rows'] = trimmed
dec_ids = sorted({d.get('id') for d in active if isinstance(d.get('id'), int)}) if decisions_block['ok'] else []
alert_ids = (sorted({a.get('id') for a in alert_rows_raw if isinstance(a.get('id'), int)})
if alerts_block['ok'] else [])
raw = (','.join(str(i) for i in dec_ids) + '|' + ','.join(str(i) for i in alert_ids) + '|' +
('1' if decisions_block['ok'] else '0') + ('1' if alerts_block['ok'] else '0'))
version = hashlib.sha1(raw.encode('utf-8')).hexdigest()[:16]
req_version = request.args.get('version', '')
if req_version and req_version == version:
return jsonify({'version': version, 'unchanged': True})
return jsonify({'version': version, 'decisions': decisions_block, 'alerts': alerts_block})
@app.route('/api/crowdsec/alerts')
@login_required
def api_cs_alerts():
lapi = _cs_lapi_url()
if not (lapi and (_cs_api_key() or _cs_has_machine())):
return jsonify({'error': gettext('CrowdSec not configured')}), 503
force_full = request.args.get('full') in ('1', 'true', 'yes')
_limit = cs_alert_limit()
try:
alerts, _mode = _crowd.cs_alerts(_limit, force_full=force_full)
except CrowdSecUnavailable as e:
status = getattr(e, 'status', 0) or 0
return jsonify({'error': _i18n.shown_error(e)}), (status if status >= 400 else 502)
except Exception as e:
logger.exception("CrowdSec alerts error")
return jsonify({'error': _i18n.shown_error(e)}), 500
out = jsonify(alerts)
out.headers['X-CS-Alert-Limit'] = str(_limit)
out.headers['X-CS-Alert-Capped'] = '1' if (_limit and len(alerts) >= _limit) else '0'
return out
@app.route('/api/crowdsec/decisions', methods=['POST'])
@csrf_protect
@login_required
def api_cs_add_decision():
lapi = _cs_lapi_url()
key = _cs_api_key()
if not (lapi and (key or _cs_has_machine())):
return jsonify({'error': gettext('CrowdSec not configured')}), 503
data = request.get_json() or {}
ip = data.get('value', '').strip()
dtype = data.get('type', 'ban').strip()
duration = data.get('duration', '24h').strip()
reason = (data.get('reason', '') or '').strip() or 'manual ban from Traefik Manager'
if not ip:
return jsonify({'error': gettext('IP/Range is required')}), 400
if dtype not in ('ban', 'captcha', 'bypass'):
return jsonify({'error': gettext('Invalid type')}), 400
now = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
payload = [{
'capacity': 0,
'decisions': [{'duration': duration, 'origin': 'manual', 'scenario': reason,
'scope': 'Ip', 'type': dtype, 'value': ip, 'simulated': False}],
'events': [], 'events_count': 1, 'labels': None, 'leakspeed': '0',
'message': reason, 'scenario': reason, 'scenario_hash': '', 'scenario_version': '',
'simulated': False,
'source': {'ip': ip, 'scope': 'Ip', 'value': ip},
'start_at': now, 'stop_at': now,
}]
if _cs_has_machine():
result = _cs_machine_request('POST', '/v1/alerts', json=payload)
else:
result = _cs_request('POST', '/v1/alerts', lapi=lapi, key=key, json=payload)
if result is None:
return jsonify({'error': gettext('Failed to add decision - check LAPI permissions')}), 502
_crowd.cs_stream_reset()
_crowd.cs_alerts_reset()
return jsonify({'ok': True})
@app.route('/api/crowdsec/decisions/<int:decision_id>', methods=['DELETE'])
@csrf_protect
@login_required
def api_cs_unban(decision_id):
if not (_cs_lapi_url() and (_cs_api_key() or _cs_has_machine())):
return jsonify({'error': gettext('CrowdSec not configured')}), 503
if _cs_has_machine():
result = _cs_machine_request('DELETE', f'/v1/decisions/{decision_id}')
else:
result = _cs_request('DELETE', f'/v1/decisions/{decision_id}')
if result is None:
return jsonify({'error': gettext('Failed to delete decision')}), 500
_crowd.cs_stream_reset()
add_notification('success', _i18n.lazy_gettext('Decision %(id)s deleted (IP unbanned)', id=decision_id),
category='crowdsec')
return jsonify({'ok': True})
@app.route('/api/ping')
@login_required
def api_route_ping():
from urllib.parse import urlparse
url = request.args.get('url', '').strip()
fallback = request.args.get('fallback', '').strip()
if not url or not url.startswith(('http://', 'https://')):
return jsonify({'ok': False, 'error': gettext('Invalid URL')}), 400
if not _ssrf_ok(url):
return jsonify({'ok': False, 'error': gettext('Target address not allowed')}), 400
host = urlparse(url).hostname or ''
tm_host = request.host.split(':')[0].lower()
if host.lower() == tm_host:
return jsonify({'ok': True, 'latency_ms': 0, 'status_code': 200, 'self': True})
settings = load_settings()
self_domain = (settings.get('self_route') or {}).get('domain', '').strip().lower()
if self_domain and host.lower() == self_domain:
return jsonify({'ok': True, 'latency_ms': 0, 'status_code': 200, 'self': True})
pool = _reach.pool_health(request.args.getlist('servers'), ssrf=_ssrf_ok, head=requests.head)
if pool:
return jsonify(_reach.pool_result(pool))
return jsonify(_reach.probe(url, fallback, ssrf=_ssrf_ok, head=requests.head))
def _apr1_hash(password: str, salt: str) -> str:
import hashlib
pw = password.encode('latin-1')
sl = salt.encode('ascii')
mgc = b'$apr1$'
a = hashlib.md5(pw + mgc + sl)
b = hashlib.md5(pw + sl + pw).digest()
plen = len(pw)
ndig, nrem = divmod(plen, 16)
for n in ndig * [16] + [nrem]:
a.update(b[:n])
i = plen
while i:
a.update(b'\x00' if (i & 1) else pw[:1])
i >>= 1
a = a.digest()
for i in range(1000):
c = hashlib.md5()
c.update(pw if (i & 1) else a)
if i % 3: c.update(sl)
if i % 7: c.update(pw)
c.update(a if (i & 1) else pw)
a = c.digest()
t64 = './0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'
def to64(v, n):
r = ''
for _ in range(n):
r += t64[v & 0x3f]; v >>= 6
return r
enc = to64((a[0]<<16)|(a[6]<<8)|a[12], 4)
enc += to64((a[1]<<16)|(a[7]<<8)|a[13], 4)
enc += to64((a[2]<<16)|(a[8]<<8)|a[14], 4)
enc += to64((a[3]<<16)|(a[9]<<8)|a[15], 4)
enc += to64((a[4]<<16)|(a[10]<<8)|a[5], 4)
enc += to64(a[11], 2)
return f'$apr1${salt}${enc}'
@app.route('/api/traefik/ping')
@login_required
def api_ping():
import time as _t
settings = load_settings()
u = settings.get('traefik_api_user', '')
p = settings.get('traefik_api_password', '')
auth = (u, p) if u and p else None
try:
t0 = _t.monotonic()
resp = requests.get(f"{settings['traefik_api_url']}/ping", timeout=3, auth=auth, verify=_traefik_verify())
ms = round((_t.monotonic() - t0) * 1000)
if resp.status_code == 200:
return jsonify({'ok': True, 'latency_ms': ms})
except Exception as e:
logger.debug(f"Ping failed: {e}")
return jsonify({'ok': False, 'latency_ms': None}), 503
@app.route('/api/manager/version')
@login_required
def api_manager_version():
mgr = _updates.release_info(GITHUB_REPO)
tfk = _updates.release_info(_updates.TRAEFIK_REPO)
return jsonify({
"version": APP_VERSION,
"repo": GITHUB_REPO,
"static_config_configured": bool(_get_static_config_path()),
"latest": mgr.get('tag', ''),
"release_url": mgr.get('url', ''),
"release_notes": mgr.get('notes', ''),
"release_error": _i18n.shown(mgr.get('error', '')),
"traefik_latest": tfk.get('tag', ''),
"traefik_release_url": tfk.get('url', ''),
"traefik_running": _updates.running_traefik_version(),
})
@app.route('/static/manifest.json')
def static_manifest():
body = render_template('manifest.json')
return app.response_class(body, mimetype='application/manifest+json')
_storage_probe_cache = {'at': 0.0, 'problems': []}
STORAGE_PROBE_TTL = 30
@app.route('/api/storage/status')
@login_required
def api_storage_status():
now = time.time()
if now - _storage_probe_cache['at'] >= STORAGE_PROBE_TTL:
_storage_probe_cache['problems'] = [
{'label': label, 'path': path, 'error': err}
for label, path, err in env.unwritable_storage()
]
_storage_probe_cache['at'] = now
return jsonify({'problems': _storage_probe_cache['problems']})
@app.route('/api/health')
def api_health():
return jsonify({"ok": True}), 200
@app.route('/api')
def api_docs():
from flask import send_from_directory
return send_from_directory('static', 'api.html')
@app.route('/openapi.yaml')
def openapi_yaml():
import re as _re
try:
with open(os.path.join(app.static_folder, 'openapi.yaml'), 'r') as f:
spec = f.read()
spec = _re.sub(r'(?m)^(\s*version:\s*).*$', rf'\g<1>{APP_VERSION}', spec, count=1)
return spec, 200, {'Content-Type': 'application/yaml'}
except Exception:
from flask import send_from_directory
return send_from_directory('static', 'openapi.yaml')
def _restart_via_docker() -> bool:
try:
import docker as _docker
client = _docker.from_env()
container = client.containers.get(_get_traefik_container())
container.restart()
logger.info(f"Restarted Traefik container: {_get_traefik_container()!r}")
return True
except Exception as e:
logger.error(f"Docker restart failed: {e}")
return False
def _restart_via_signal_file() -> bool:
try:
sig_path = _get_signal_file_path()
os.makedirs(os.path.dirname(sig_path), exist_ok=True)
with open(sig_path, 'w') as f:
f.write('')
logger.info(f"Restart signal written: {sig_path}")
return True
except Exception as e:
logger.error(f"Signal file write failed: {e}")
return False
def trigger_traefik_restart() -> tuple:
method = _get_restart_method()
if method in ('proxy', 'socket'):
ok = _restart_via_docker()
return ok, ('' if ok else gettext('Docker restart failed - check DOCKER_HOST and TRAEFIK_CONTAINER (%(container)s)',
container=repr(_get_traefik_container())))
if method == 'poison-pill':
ok = _restart_via_signal_file()
return ok, ('' if ok else gettext('Signal file write failed - check SIGNAL_FILE_PATH (%(path)s)',
path=repr(_get_signal_file_path())))
return False, gettext('Unknown RESTART_METHOD: %(method)s', method=repr(method))
@app.route('/api/static/available')
@login_required
def api_static_available():
path = _get_static_config_path()
return jsonify({'available': bool(path and os.path.exists(path))})
@app.route('/api/static/config')
@login_required
def api_static_config_get():
server = (request.args.get('server') or '').strip()
if server:
agent = _agent_by_id(server)
if not agent:
return jsonify({'error': gettext('Agent not found')}), 404
try:
resp = _agent_request(agent, 'GET', '/api/static')
except requests.exceptions.RequestException as e:
return jsonify({'error': gettext('Cannot reach agent: %(error)s', error=e)}), 502
if resp.status_code != 200:
try:
msg = _agent_err.localize(resp.json() or {}).get('error', '')
except Exception:
msg = ''
return jsonify({'error': msg or gettext('Static config not available on this agent')}), resp.status_code
body = resp.json() or {}
raw = body.get('content', '')
try:
_y = SafeYAML(typ='safe')
parsed = _y.load(raw) or {}
except Exception as e:
return jsonify({'error': gettext('Agent static config is not valid YAML: %(error)s', error=e)}), 500
return jsonify({'raw': raw, 'parsed': parsed, 'path': body.get('path', '')})
path = _readable_config_path(_get_static_config_path())
if not path or not os.path.exists(path):
refused = _settings.refused_path('static')
if refused:
return jsonify({'error': gettext('Static config path refused: %(refused)s', refused=_i18n.shown(refused))}), 404
return jsonify({'error': gettext('Static config not found or STATIC_CONFIG_PATH not set')}), 404
try:
with open(path, 'r') as f:
raw = f.read()
_y = SafeYAML(typ='safe')
parsed = _y.load(raw) or {}
return jsonify({'raw': raw, 'parsed': parsed, 'path': path})
except Exception as e:
logger.exception("Failed to read static config")
return jsonify({'error': str(e)}), 500
@app.route('/api/static/config', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_static_config_save():
path = _get_static_config_path()
if not path:
refused = _settings.refused_path('static')
if refused:
return jsonify({'error': gettext('Static config path refused: %(refused)s', refused=_i18n.shown(refused))}), 400
return jsonify({'error': gettext('STATIC_CONFIG_PATH not configured')}), 400
safe_path = _safe_file_path(path)
if not safe_path:
return jsonify({'error': gettext('Static config path is outside allowed directories')}), 403
data = request.get_json(silent=True) or {}
content = (data.get('content') or data.get('raw') or '').strip()
if not content:
return jsonify({'error': gettext('No content provided')}), 400
try:
_y = SafeYAML(typ='safe')
_new_doc = _y.load(content)
except Exception as e:
return jsonify({'error': gettext('Invalid YAML: %(error)s', error=e)}), 400
_before = {}
try:
if os.path.exists(safe_path):
with open(safe_path) as _fh:
_before = SafeYAML(typ='safe').load(_fh.read()) or {}
except Exception:
_before = {}
_renames, _gone = _plugin_diff(_static_plugins(_before), _static_plugins(_new_doc))
_dyn = [load_config(_p) for _p in env.CONFIG_PATHS]
for _name in _gone:
_users = _middlewares_using_plugin(_dyn, _name)
if _users:
return jsonify({'error': gettext('%(name)s is still used by %(users)s. Delete those middlewares first', name=_name, users=_name_list(_users)),
'inUseBy': _users}), 409
try:
create_backup(safe_path)
with open(safe_path, 'w') as f:
f.write(content)
for _old, _new in _renames.items():
_cascade_across_configs(None, lambda c, o=_old, n=_new: _retarget_plugin(c, o, n))
logger.info(f"Static config saved by {request.remote_addr}: {safe_path}")
add_notification('success', _i18n.lazy_gettext('Static config saved'))
threading.Thread(target=lambda: _git_push_if_enabled('static config save'), daemon=True).start()
return jsonify({'ok': True})
except Exception as e:
logger.exception("Failed to write static config")
return jsonify({'error': str(e)}), 500
@app.route('/api/static/restart', methods=['POST'])
@csrf_protect
@login_required
def api_static_restart():
ok, err = trigger_traefik_restart()
if ok:
logger.info(f"Traefik restarted via static config by {request.remote_addr}")
add_notification('warning', _i18n.lazy_gettext('Traefik restarted'), category='traefik')
return jsonify({'ok': True})
logger.error(f"Traefik restart failed for {request.remote_addr} (RESTART_METHOD {_get_restart_method()!r})")
return jsonify({'ok': False, 'error': err}), 500
@app.route('/api/static/status')
@login_required
def api_static_status():
data = traefik_api_get('/api/overview')
return jsonify({'up': data is not None})
@app.route('/api/traefik/runtime')
@login_required
def api_traefik_runtime():
method = _get_restart_method()
container = _get_traefik_container()
if method in ('proxy', 'socket'):
return jsonify({'method': method, 'runtime': 'docker', 'container': container})
if method == 'poison-pill':
try:
import docker as _docker
client = _docker.from_env()
client.containers.get(container)
return jsonify({'method': method, 'runtime': 'docker', 'container': container})
except Exception:
return jsonify({'method': method, 'runtime': 'native', 'container': None})
return jsonify({'method': method, 'runtime': 'unknown', 'container': None})
@app.route('/api/static/section', methods=['POST'])
@csrf_protect
@login_required
def api_static_section_update():
req = request.get_json(silent=True) or {}
path = _get_static_config_path()
if not req.get('current_raw') and (not path or not os.path.exists(path)):
return jsonify({'error': gettext('Static config not found')}), 404
action = req.get('action', '')
section = req.get('section', '')
name = str(req.get('name', '')).strip()
old_name = str(req.get('old_name', name)).strip()
payload = req.get('data', {})
if not action or not section or (action not in ('set', 'remove') and not name):
return jsonify({'error': gettext('Missing required fields')}), 400
current_raw = req.get('current_raw', '')
try:
_y = YAML()
_y.preserve_quotes = True
if current_raw:
config = _y.load(StringIO(current_raw)) or {}
else:
with open(path, 'r') as f:
config = _y.load(f) or {}
if section == 'entrypoints':
eps = config.setdefault('entryPoints', {})
if action == 'remove':
eps.pop(name, None)
else:
if action == 'edit' and old_name != name:
eps[name] = eps.pop(old_name, None)
ep = eps.get(name)
if not isinstance(ep, dict):
ep = {}
addr = str(payload.get('address', '')).strip()
if addr:
ep['address'] = DoubleQuotedScalarString(addr)
elif 'address' not in ep:
ep['address'] = ''
http_blk = ep.get('http') if isinstance(ep.get('http'), dict) else {}
redirect_to = str(payload.get('redirect_to', '')).strip()
if redirect_to:
http_blk['redirections'] = {'entryPoint': {'to': redirect_to, 'scheme': 'https', 'permanent': True}}
else:
http_blk.pop('redirections', None)
uhs = str(payload.get('underscore_headers', '')).strip().lower()
hdr_key = str(payload.get('headers_strategy_key', '')).strip()
if hdr_key not in ('aliasHeadersStrategy', 'underscoreHeadersStrategy'):
if 'aliasHeadersStrategy' in http_blk:
hdr_key = 'aliasHeadersStrategy'
uhs = uhs or str(http_blk['aliasHeadersStrategy']).strip().lower()
else:
hdr_key = 'underscoreHeadersStrategy'
http_blk.pop('aliasHeadersStrategy', None)
http_blk.pop('underscoreHeadersStrategy', None)
if uhs in ('keep', 'delete', 'reject'):
http_blk[hdr_key] = uhs
mws = [m for m in re.split(r'[\s,]+', str(payload.get('middlewares', ''))) if m]
if mws:
http_blk['middlewares'] = mws
else:
http_blk.pop('middlewares', None)
if payload.get('tls_enabled'):
tls_blk = http_blk.get('tls') if isinstance(http_blk.get('tls'), dict) else {}
cr = str(payload.get('tls_cert_resolver', '')).strip()
if cr:
tls_blk['certResolver'] = cr
else:
tls_blk.pop('certResolver', None)
topt = str(payload.get('tls_options', '')).strip()
if topt:
tls_blk['options'] = topt
else:
tls_blk.pop('options', None)
http_blk['tls'] = tls_blk
else:
http_blk.pop('tls', None)
if http_blk:
ep['http'] = http_blk
else:
ep.pop('http', None)
if payload.get('http3'):
if not isinstance(ep.get('http3'), dict):
ep['http3'] = {}
else:
ep.pop('http3', None)
if payload.get('as_default'):
ep['asDefault'] = True
else:
ep.pop('asDefault', None)
tr = ep.get('transport') if isinstance(ep.get('transport'), dict) else {}
rts = tr.get('respondingTimeouts') if isinstance(tr.get('respondingTimeouts'), dict) else {}
for yaml_key, pay_key in (('readTimeout', 'read_timeout'), ('writeTimeout', 'write_timeout'), ('idleTimeout', 'idle_timeout')):
v = str(payload.get(pay_key, '')).strip()
if v:
if not _is_valid_duration(v):
return jsonify({'error': gettext('Invalid duration for %(yaml_key)s: %(value)s - use forms like 30, 30s, 1m30s', yaml_key=yaml_key, value=repr(v))}), 400
rts[yaml_key] = int(v) if v.isdigit() else v
else:
rts.pop(yaml_key, None)
if rts:
tr['respondingTimeouts'] = rts
else:
tr.pop('respondingTimeouts', None)
if tr:
ep['transport'] = tr
else:
ep.pop('transport', None)
fwd_ips = _parse_cidr_input(payload.get('trusted_ips'))
pp_ips = _parse_cidr_input(payload.get('proxy_trusted_ips'))
bad = [c for c in fwd_ips + pp_ips if not _is_valid_cidr(c)]
if bad:
return jsonify({'error': gettext('Invalid IP/CIDR: %(bad)s', bad=', '.join(bad))}), 400
fh = ep.get('forwardedHeaders') if isinstance(ep.get('forwardedHeaders'), dict) else {}
if fwd_ips:
fh['trustedIPs'] = fwd_ips
else:
fh.pop('trustedIPs', None)
if payload.get('forwarded_insecure'):
fh['insecure'] = True
else:
fh.pop('insecure', None)
if fh:
ep['forwardedHeaders'] = fh
else:
ep.pop('forwardedHeaders', None)
pp = ep.get('proxyProtocol') if isinstance(ep.get('proxyProtocol'), dict) else {}
if pp_ips:
pp['trustedIPs'] = pp_ips
else:
pp.pop('trustedIPs', None)
if payload.get('proxy_insecure'):
pp['insecure'] = True
else:
pp.pop('insecure', None)
if pp:
ep['proxyProtocol'] = pp
else:
ep.pop('proxyProtocol', None)
eps[name] = ep
elif section == 'resolvers':
resolvers = config.setdefault('certificatesResolvers', {})
if action == 'remove':
resolvers.pop(name, None)
else:
if action == 'edit' and old_name != name:
existing_res = resolvers.pop(old_name, None)
else:
existing_res = resolvers.get(name)
if not isinstance(existing_res, dict):
existing_res = {}
acme = existing_res.get('acme') if isinstance(existing_res.get('acme'), dict) else {}
acme['email'] = payload.get('email', '')
acme['storage'] = payload.get('storage', '/acme.json')
ct = payload.get('challenge_type', 'dnsChallenge')
if ct == 'dnsChallenge':
dns = acme.get('dnsChallenge') if isinstance(acme.get('dnsChallenge'), dict) else {}
dns['provider'] = payload.get('provider', '')
dns_res = [r for r in re.split(r'[\s,]+', str(payload.get('dns_resolvers', ''))) if r]
if dns_res:
dns['resolvers'] = dns_res
else:
dns.pop('resolvers', None)
prop = dns.get('propagation') if isinstance(dns.get('propagation'), dict) else {}
delay = str(payload.get('dns_delay', '')).strip()
if delay:
if not _is_valid_duration(delay):
return jsonify({'error': gettext('Invalid propagation delay: %(delay)s - use forms like 30, 30s, 2m', delay=repr(delay))}), 400
prop['delayBeforeChecks'] = int(delay) if delay.isdigit() else delay
else:
prop.pop('delayBeforeChecks', None)
if payload.get('dns_disable_checks'):
prop['disableChecks'] = True
else:
prop.pop('disableChecks', None)
if prop:
dns['propagation'] = prop
else:
dns.pop('propagation', None)
acme['dnsChallenge'] = dns
acme.pop('httpChallenge', None)
acme.pop('tlsChallenge', None)
elif ct == 'httpChallenge':
http_ch = acme.get('httpChallenge') if isinstance(acme.get('httpChallenge'), dict) else {}
http_ch['entryPoint'] = payload.get('http_entrypoint', 'web')
acme['httpChallenge'] = http_ch
acme.pop('dnsChallenge', None)
acme.pop('tlsChallenge', None)
else:
acme.setdefault('tlsChallenge', {})
acme.pop('dnsChallenge', None)
acme.pop('httpChallenge', None)
ca = str(payload.get('ca_server', '')).strip()
if ca:
acme['caServer'] = ca
else:
acme.pop('caServer', None)
kt = str(payload.get('key_type', '')).strip()
if kt:
acme['keyType'] = kt
else:
acme.pop('keyType', None)
eab_kid = str(payload.get('eab_kid', '')).strip()
eab_hmac = str(payload.get('eab_hmac', '')).strip()
if eab_kid and eab_hmac:
acme['eab'] = {'kid': eab_kid, 'hmacEncoded': eab_hmac}
elif eab_kid or eab_hmac:
return jsonify({'error': gettext('EAB needs both the key ID and the HMAC')}), 400
else:
acme.pop('eab', None)
existing_res['acme'] = acme
resolvers[name] = existing_res
elif section == 'plugins':
exp = config.setdefault('experimental', {})
plugins = exp.get('plugins') if isinstance(exp.get('plugins'), dict) else {}
local = exp.get('localPlugins') if isinstance(exp.get('localPlugins'), dict) else {}
if action == 'remove':
plugins.pop(name, None)
local.pop(name, None)
else:
if action == 'edit' and old_name != name:
plugins.pop(old_name, None)
local.pop(old_name, None)
if payload.get('local'):
plugins.pop(name, None)
local[name] = {'moduleName': payload.get('moduleName', '')}
else:
local.pop(name, None)
plugins[name] = {'moduleName': payload.get('moduleName', ''), 'version': payload.get('version', '')}
if plugins:
exp['plugins'] = plugins
else:
exp.pop('plugins', None)
if local:
exp['localPlugins'] = local
else:
exp.pop('localPlugins', None)
if not exp:
config.pop('experimental', None)
elif section == 'api' and action == 'set':
if payload.get('enabled', True):
api_cfg = config.get('api')
if not isinstance(api_cfg, dict):
api_cfg = {}
if not payload.get('dashboard', True):
api_cfg['dashboard'] = False
elif 'dashboard' in api_cfg:
api_cfg['dashboard'] = True
if payload.get('insecure'):
api_cfg['insecure'] = True
elif 'insecure' in api_cfg:
api_cfg['insecure'] = False
if payload.get('debug'):
api_cfg['debug'] = True
else:
api_cfg.pop('debug', None)
config['api'] = api_cfg
else:
config.pop('api', None)
elif section == 'log' and action == 'set':
log_blk = config.get('log') if isinstance(config.get('log'), dict) else {}
level = str(payload.get('level', 'ERROR')).upper()
if level and level != 'ERROR':
log_blk['level'] = level
else:
log_blk.pop('level', None)
if str(payload.get('log_format', '')).strip() == 'json':
log_blk['format'] = 'json'
else:
log_blk.pop('format', None)
log_file = str(payload.get('log_file', '')).strip()
if log_file:
log_blk['filePath'] = log_file
else:
log_blk.pop('filePath', None)
for yaml_key, pay_key in (('maxSize', 'log_max_size'), ('maxBackups', 'log_max_backups'), ('maxAge', 'log_max_age')):
v = str(payload.get(pay_key, '')).strip()
if v and log_file:
if not v.isdigit():
return jsonify({'error': gettext('Rotation %(yaml_key)s must be a whole number, got %(value)s', yaml_key=yaml_key, value=repr(v))}), 400
log_blk[yaml_key] = int(v)
else:
log_blk.pop(yaml_key, None)
if payload.get('log_compress') and log_file:
log_blk['compress'] = True
else:
log_blk.pop('compress', None)
if log_blk:
config['log'] = log_blk
else:
config.pop('log', None)
if payload.get('accessLog'):
al = config.get('accessLog') if isinstance(config.get('accessLog'), dict) else {}
al_path = str(payload.get('accessLogPath', '')).strip()
if al_path:
al['filePath'] = al_path
else:
al.pop('filePath', None)
if str(payload.get('al_format', '')).strip() == 'json':
al['format'] = 'json'
else:
al.pop('format', None)
buf = str(payload.get('al_buffering', '')).strip()
if buf:
if not buf.isdigit():
return jsonify({'error': gettext('Buffering must be a whole number of lines, got %(buffer)s', buffer=repr(buf))}), 400
al['bufferingSize'] = int(buf)
else:
al.pop('bufferingSize', None)
filters = al.get('filters') if isinstance(al.get('filters'), dict) else {}
codes = [c for c in re.split(r'[\s,]+', str(payload.get('al_status_codes', ''))) if c]
if codes:
bad_codes = [c for c in codes if not re.match(r'^\d{3}(-\d{3})?$', c)]
if bad_codes:
return jsonify({'error': gettext('Invalid status code filter: %(bad_codes)s', bad_codes=', '.join(bad_codes))}), 400
filters['statusCodes'] = codes
else:
filters.pop('statusCodes', None)
if payload.get('al_retry'):
filters['retryAttempts'] = True
else:
filters.pop('retryAttempts', None)
min_dur = str(payload.get('al_min_duration', '')).strip()
if min_dur:
if not _is_valid_duration(min_dur):
return jsonify({'error': gettext('Invalid min duration: %(min_dur)s - use forms like 200ms, 1s', min_dur=repr(min_dur))}), 400
filters['minDuration'] = int(min_dur) if min_dur.isdigit() else min_dur
else:
filters.pop('minDuration', None)
if filters:
al['filters'] = filters
else:
al.pop('filters', None)
hdr_mode = str(payload.get('al_headers_mode', '')).strip()
fields = al.get('fields') if isinstance(al.get('fields'), dict) else {}
hdrs = fields.get('headers') if isinstance(fields.get('headers'), dict) else {}
if hdr_mode in ('keep', 'redact'):
hdrs['defaultMode'] = hdr_mode
else:
hdrs.pop('defaultMode', None)
if hdrs:
fields['headers'] = hdrs
else:
fields.pop('headers', None)
if fields:
al['fields'] = fields
else:
al.pop('fields', None)
config['accessLog'] = al
else:
config.pop('accessLog', None)
elif section == 'providers' and action == 'set':
providers = config.setdefault('providers', {})
if payload.get('docker'):
_existing = providers.get('docker')
docker_cfg = _existing if isinstance(_existing, dict) else {}
endpoint = str(payload.get('dockerEndpoint', '')).strip()
if endpoint and endpoint != 'unix:///var/run/docker.sock':
docker_cfg['endpoint'] = endpoint
else:
docker_cfg.pop('endpoint', None)
if not payload.get('dockerExposedByDefault', True):
docker_cfg['exposedByDefault'] = False
else:
docker_cfg.pop('exposedByDefault', None)
if not payload.get('dockerWatch', True):
docker_cfg['watch'] = False
else:
docker_cfg.pop('watch', None)
providers['docker'] = docker_cfg
else:
providers.pop('docker', None)
if payload.get('file'):
_existing = providers.get('file')
file_cfg = _existing if isinstance(_existing, dict) else {}
directory = str(payload.get('fileDirectory', '')).strip()
if directory:
file_cfg['directory'] = directory
else:
file_cfg.pop('directory', None)
if not payload.get('fileWatch', True):
file_cfg['watch'] = False
else:
file_cfg.pop('watch', None)
providers['file'] = file_cfg
else:
providers.pop('file', None)
throttle = str(payload.get('providers_throttle', '')).strip()
if throttle:
if not _is_valid_duration(throttle):
return jsonify({'error': gettext('Invalid providers throttle: %(throttle)s - use forms like 2s, 500ms', throttle=repr(throttle))}), 400
providers['providersThrottleDuration'] = int(throttle) if throttle.isdigit() else throttle
else:
providers.pop('providersThrottleDuration', None)
if not providers:
config.pop('providers', None)
elif section == 'providers' and action in ('add', 'edit', 'remove'):
providers = config.setdefault('providers', {})
if action == 'remove':
providers.pop(name, None)
else:
if action == 'edit' and old_name and old_name != name:
providers.pop(old_name, None)
prov_cfg = {}
yaml_config = str(payload.get('yaml_config', '')).strip()
if yaml_config:
try:
_yp = SafeYAML(typ='safe')
parsed = _yp.load(yaml_config)
if isinstance(parsed, dict):
prov_cfg = parsed
except Exception:
pass
providers[name] = prov_cfg
if not providers:
config.pop('providers', None)
elif section == 'observability' and action == 'set':
if payload.get('ping'):
if not isinstance(config.get('ping'), dict):
config['ping'] = {}
else:
config.pop('ping', None)
metrics = config.get('metrics') if isinstance(config.get('metrics'), dict) else {}
if payload.get('prometheus'):
prom = metrics.get('prometheus') if isinstance(metrics.get('prometheus'), dict) else {}
if payload.get('prom_ep_labels', True):
prom.pop('addEntryPointsLabels', None)
else:
prom['addEntryPointsLabels'] = False
if payload.get('prom_router_labels'):
prom['addRoutersLabels'] = True
else:
prom.pop('addRoutersLabels', None)
if payload.get('prom_svc_labels', True):
prom.pop('addServicesLabels', None)
else:
prom['addServicesLabels'] = False
metrics['prometheus'] = prom
else:
metrics.pop('prometheus', None)
if metrics:
config['metrics'] = metrics
else:
config.pop('metrics', None)
if payload.get('tracing'):
tr_blk = config.get('tracing') if isinstance(config.get('tracing'), dict) else {}
svc = str(payload.get('trace_service', '')).strip()
if svc:
tr_blk['serviceName'] = svc
else:
tr_blk.pop('serviceName', None)
sr = str(payload.get('trace_sample', '')).strip()
if sr:
try:
srf = float(sr)
if not 0 <= srf <= 1:
raise ValueError
except ValueError:
return jsonify({'error': gettext('Sample rate must be a number between 0 and 1, got %(route)s', route=repr(sr))}), 400
tr_blk['sampleRate'] = srf
else:
tr_blk.pop('sampleRate', None)
ep_url = str(payload.get('trace_endpoint', '')).strip()
otlp = tr_blk.get('otlp') if isinstance(tr_blk.get('otlp'), dict) else {}
if ep_url:
oh = otlp.get('http') if isinstance(otlp.get('http'), dict) else {}
oh['endpoint'] = ep_url
otlp['http'] = oh
else:
oh = otlp.get('http') if isinstance(otlp.get('http'), dict) else None
if oh is not None:
oh.pop('endpoint', None)
if not oh:
otlp.pop('http', None)
if otlp:
tr_blk['otlp'] = otlp
else:
tr_blk.pop('otlp', None)
config['tracing'] = tr_blk
else:
config.pop('tracing', None)
elif section == 'system' and action == 'set':
g_blk = config.get('global') if isinstance(config.get('global'), dict) else {}
if payload.get('check_new_version', True):
g_blk.pop('checkNewVersion', None)
else:
g_blk['checkNewVersion'] = False
if payload.get('send_usage'):
g_blk['sendAnonymousUsage'] = True
else:
g_blk.pop('sendAnonymousUsage', None)
if 'global_alias_headers' in payload:
global_alias = str(payload.get('global_alias_headers') or '').strip().lower()
if global_alias not in ('', 'keep', 'delete', 'reject'):
return jsonify({'error': gettext('Alias Headers must be keep, delete or reject, got %(value)s', value=global_alias)}), 400
if global_alias:
g_blk['aliasHeadersStrategy'] = global_alias
else:
g_blk.pop('aliasHeadersStrategy', None)
if g_blk:
config['global'] = g_blk
else:
config.pop('global', None)
core_blk = config.get('core') if isinstance(config.get('core'), dict) else {}
if str(payload.get('rule_syntax', '')).strip() == 'v2':
core_blk['defaultRuleSyntax'] = 'v2'
else:
core_blk.pop('defaultRuleSyntax', None)
if core_blk:
config['core'] = core_blk
else:
config.pop('core', None)
st = config.get('serversTransport') if isinstance(config.get('serversTransport'), dict) else {}
if payload.get('st_insecure'):
st['insecureSkipVerify'] = True
else:
st.pop('insecureSkipVerify', None)
cas = [c for c in re.split(r'[\s,]+', str(payload.get('st_root_cas', ''))) if c]
if cas:
st['rootCAs'] = cas
else:
st.pop('rootCAs', None)
max_idle = str(payload.get('st_max_idle', '')).strip()
if max_idle:
if not max_idle.isdigit():
return jsonify({'error': gettext('Max idle conns must be a whole number, got %(max_idle)s', max_idle=repr(max_idle))}), 400
st['maxIdleConnsPerHost'] = int(max_idle)
else:
st.pop('maxIdleConnsPerHost', None)
fwd_t = st.get('forwardingTimeouts') if isinstance(st.get('forwardingTimeouts'), dict) else {}
for yaml_key, pay_key in (('dialTimeout', 'st_dial'), ('responseHeaderTimeout', 'st_resp_header'), ('idleConnTimeout', 'st_idle_conn')):
v = str(payload.get(pay_key, '')).strip()
if v:
if not _is_valid_duration(v):
return jsonify({'error': gettext('Invalid duration for %(yaml_key)s: %(value)s', yaml_key=yaml_key, value=repr(v))}), 400
fwd_t[yaml_key] = int(v) if v.isdigit() else v
else:
fwd_t.pop(yaml_key, None)
if fwd_t:
st['forwardingTimeouts'] = fwd_t
else:
st.pop('forwardingTimeouts', None)
if st:
config['serversTransport'] = st
else:
config.pop('serversTransport', None)
else:
return jsonify({'error': gettext('Unknown section: %(section)s', section=repr(section))}), 400
stream = StringIO()
_y.dump(config, stream)
new_raw = stream.getvalue()
_y2 = SafeYAML(typ='safe')
parsed = _y2.load(new_raw) or {}
return jsonify({'ok': True, 'raw': new_raw, 'parsed': parsed})
except Exception as e:
logger.exception("Static section update failed")
return jsonify({'error': str(e)}), 500
_CLOUDFLARE_IPS_CAPTURED = '2026-07-23'
_CLOUDFLARE_IPS_V4 = [
'173.245.48.0/20', '103.21.244.0/22', '103.22.200.0/22', '103.31.4.0/22',
'141.101.64.0/18', '108.162.192.0/18', '190.93.240.0/20', '188.114.96.0/20',
'197.234.240.0/22', '198.41.128.0/17', '162.158.0.0/15', '104.16.0.0/13',
'104.24.0.0/14', '172.64.0.0/13', '131.0.72.0/22',
]
_CLOUDFLARE_IPS_V6 = [
'2400:cb00::/32', '2606:4700::/32', '2803:f800::/32', '2405:b500::/32',
'2405:8100::/32', '2a06:98c0::/29', '2c0f:f248::/32',
]
_PRIVATE_IP_RANGES = ['10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16', 'fc00::/7']
_DURATION_RE = re.compile(r'^(\d+(\.\d+)?(ns|us|µs|ms|s|m|h))+$')
def _is_valid_duration(v: str) -> bool:
v = str(v).strip()
return bool(v) and (v.isdigit() or bool(_DURATION_RE.match(v)))
def _is_valid_cidr(cidr: str) -> bool:
try:
ipaddress.ip_network(str(cidr).strip(), strict=False)
return True
except ValueError:
return False
def _parse_cidr_input(raw) -> list:
if isinstance(raw, list):
parts = [str(x) for x in raw]
else:
parts = re.split(r'[\s,]+', str(raw or ''))
return [p.strip() for p in parts if p.strip()]
@app.route('/api/static/trusted-ips/preview', methods=['POST'])
@csrf_protect
@login_required
def api_static_trusted_ips_preview():
req = request.get_json(silent=True) or {}
current_raw = req.get('current_raw', '')
entrypoint = str(req.get('entrypoint', '')).strip()
try:
_y = YAML()
_y.preserve_quotes = True
if current_raw:
config = _y.load(StringIO(current_raw)) or {}
else:
path = _get_static_config_path()
if not path or not os.path.exists(path):
return jsonify({'error': gettext('Static config not found')}), 404
with open(path, 'r') as f:
config = _y.load(f) or {}
if not isinstance(config, dict):
return jsonify({'error': gettext('Static config is not a mapping')}), 400
ep_key = 'entryPoints' if 'entryPoints' in config else ('entrypoints' if 'entrypoints' in config else 'entryPoints')
eps = config.get(ep_key)
summary = []
if isinstance(eps, dict):
for nm, cfg in eps.items():
cur = []
addr = ''
if isinstance(cfg, dict):
addr = str(cfg.get('address', ''))
fh = cfg.get('forwardedHeaders')
if isinstance(fh, dict) and isinstance(fh.get('trustedIPs'), list):
cur = [str(x) for x in fh['trustedIPs']]
summary.append({'name': str(nm), 'address': addr, 'trusted_ips': cur})
resp = {
'ok': True,
'entrypoints': summary,
'cloudflare_captured': _CLOUDFLARE_IPS_CAPTURED,
'cloudflare_ranges': _CLOUDFLARE_IPS_V4 + _CLOUDFLARE_IPS_V6,
'private_ranges': _PRIVATE_IP_RANGES,
}
if not entrypoint:
return jsonify(resp)
if not isinstance(eps, dict) or entrypoint not in eps:
return jsonify({'error': gettext('Entrypoint "%(entrypoint)s" not found in static config', entrypoint=entrypoint)}), 400
custom = _parse_cidr_input(req.get('custom_cidrs', []))
invalid = [c for c in custom if not _is_valid_cidr(c)]
additions = []
if req.get('cloudflare'):
additions += _CLOUDFLARE_IPS_V4 + _CLOUDFLARE_IPS_V6
if req.get('private'):
additions += _PRIVATE_IP_RANGES
additions += [c for c in custom if _is_valid_cidr(c)]
ep = eps.get(entrypoint)
if not isinstance(ep, dict):
ep = {}
fh = ep.get('forwardedHeaders') if isinstance(ep.get('forwardedHeaders'), dict) else {}
cur_seq = fh.get('trustedIPs')
existing = [str(x) for x in cur_seq] if isinstance(cur_seq, list) else []
final, added = _merge_trusted_ips(existing, additions)
if isinstance(cur_seq, list):
for c in added:
cur_seq.append(DoubleQuotedScalarString(c))
else:
fh['trustedIPs'] = [DoubleQuotedScalarString(c) for c in final]
ep['forwardedHeaders'] = fh
eps[entrypoint] = ep
stream = StringIO()
_y.dump(config, stream)
new_raw = stream.getvalue()
parsed = SafeYAML(typ='safe').load(new_raw) or {}
resp.update({
'entrypoint': entrypoint,
'existing': existing,
'added': added,
'final': final,
'invalid': invalid,
'raw': new_raw,
'parsed': parsed,
})
return jsonify(resp)
except Exception as e:
logger.exception("Trusted IPs preview failed")
return jsonify({'error': str(e)}), 500
@app.route('/api/tools/digestauth', methods=['POST'])
@login_required
def api_digestauth():
import hashlib
data = request.get_json(silent=True) or {}
username = data.get('username', '').strip()
realm = data.get('realm', '').strip()
password = data.get('password', '')
if not username or not realm or not password:
return jsonify({'ok': False, 'error': gettext('username, realm and password required')}), 400
h = hashlib.md5(f'{username}:{realm}:{password}'.encode()).hexdigest()
return jsonify({'ok': True, 'hash': f'{username}:{realm}:{h}'})
@app.route('/api/tools/htpasswd', methods=['POST'])
@login_required
def api_htpasswd():
import random, string
data = request.get_json(silent=True) or {}
username = data.get('username', '').strip()
password = data.get('password', '')
if not username or not password:
return jsonify({'ok': False, 'error': gettext('username and password required')}), 400
salt = ''.join(random.choices(string.ascii_letters + string.digits + './', k=8))
h = _apr1_hash(password, salt)
return jsonify({'ok': True, 'hash': f'{username}:{h}'})
@app.route('/api/geoip/status')
@login_required
def api_geoip_status():
return jsonify(_geoip_status())
@app.route('/api/geoip/lookup', methods=['POST'])
@csrf_protect
@login_required
def api_geoip_lookup():
if not _geoip_enabled():
return jsonify({'enabled': False, 'available': False, 'results': {}})
data = request.get_json(silent=True) or {}
ips = data.get('ips') or []
if not isinstance(ips, list):
ips = []
reader = _geoip_reader()
available = reader is not None
aggregate = bool(data.get('aggregate'))
results = {}
counts = {}
codes = {}
if available:
seen = set()
for raw in ips:
ip = str(raw).strip()
if not ip or ip in seen:
continue
seen.add(ip)
geo = _geoip_lookup(ip, reader)
if not geo:
continue
if aggregate:
cc = geo.get('country_code')
if cc:
entry = counts.setdefault(cc, {'count': 0, 'country': geo.get('country')})
entry['count'] += 1
codes[ip] = cc
else:
results[ip] = geo
payload = {'enabled': True, 'available': available}
if aggregate:
payload['counts'] = counts
payload['codes'] = codes
else:
payload['results'] = results
return jsonify(payload)
@app.route('/api/geoip/update', methods=['POST'])
@csrf_protect
@login_required
@limiter.limit("6 per hour")
def api_geoip_update():
ok, info = _geoip_download()
if ok:
add_notification('success', _i18n.lazy_gettext('GeoIP database updated (DB-IP %(db_month)s)', db_month=info),
category='update')
return jsonify({'success': True, 'db_month': info, 'status': _geoip_status()})
return jsonify({'success': False, 'error': gettext('Download failed: %(info)s', info=info)}), 502
def _traefik_api_unreachable_error(exc, url):
verify = _reach.cert_verify_reason(exc)
if verify:
return gettext('TLS verification failed - the API certificate is not trusted. Mount your CA into /etc/ssl/certs/ca-certificates.crt or set TRAEFIK_INSECURE_SKIP_VERIFY=true. (%(error)s)', error=verify)
reason = _reach.failure_reason(exc, url) or str(exc)[:120]
return _i18n.lazy_gettext('Connection error - check the URL and network. (%(error)s)', error=reason)
@app.route('/api/setup/test-connection', methods=['POST'])
@login_required
def api_setup_test_connection():
settings = load_settings()
if settings.get('setup_complete', False):
return jsonify({'ok': False, 'error': gettext('Setup already complete')}), 403
data = request.get_json(silent=True) or {}
raw_url = str(data.get('url', '')).strip()
url = _safe_api_url(raw_url)
if not url:
return jsonify({'ok': False, 'error': gettext('Invalid URL')}), 400
if not _ssrf_ok(url):
return jsonify({'ok': False, 'error': gettext('Target address not allowed')}), 400
u = str(data.get('user', '')).strip()
p = str(data.get('password', '')).strip()
auth = (u, p) if u and p else None
try:
resp = requests.get(f"{url}/api/version", timeout=4, auth=auth,
verify=_traefik_verify(), allow_redirects=False)
if resp.status_code == 200:
info = resp.json()
return jsonify({'ok': True, 'version': info.get('Version', '?')})
if resp.status_code in (401, 403):
return jsonify({'ok': False, 'error': gettext('HTTP %(status_code)s - check the API username and password', status_code=resp.status_code)})
return jsonify({'ok': False, 'error': gettext('HTTP %(status_code)s from %(url)s/api/version', status_code=resp.status_code, url=url)})
except requests.exceptions.SSLError as e:
return jsonify({'ok': False, 'error': _traefik_api_unreachable_error(e, url)})
except requests.exceptions.Timeout:
return jsonify({'ok': False, 'error': gettext('Connection timed out - the API URL may be unreachable from the container')})
except requests.exceptions.ConnectionError as e:
return jsonify({'ok': False, 'error': _traefik_api_unreachable_error(e, url)})
except Exception as e:
return jsonify({'ok': False, 'error': str(e)[:160]})
@app.route('/api/traefik/router/<protocol>/<path:name>')
@login_required
def api_router_detail(protocol, name):
proto = {'http': 'http', 'tcp': 'tcp', 'udp': 'udp'}.get(protocol.lower(), 'http')
return jsonify(traefik_api_get(f'/api/{proto}/routers/{name}') or {})
@app.route('/api/traefik/plugins')
@login_required
def api_plugins():
static_path = _get_static_config_path()
if not os.path.exists(static_path):
return jsonify({'plugins': [], 'error': gettext('Static config not found at %(static_path)s. Set STATIC_CONFIG_PATH env var or configure the path in Settings.', static_path=static_path)})
try:
with open(static_path, 'r') as f:
static = yaml.load(f) or {}
raw = (static.get('experimental') or {}).get('plugins') or {}
plugins = [
{'name': n, 'moduleName': i.get('moduleName',''), 'version': i.get('version',''), 'settings': i.get('settings')}
for n, i in raw.items() if isinstance(i, dict)
]
return jsonify({'plugins': plugins})
except Exception as e:
logger.exception("Error reading static config")
return jsonify({'plugins': [], 'error': str(e)})
_PLUGIN_CATALOG = {'ts': 0.0, 'map': {}}
@app.route('/api/plugins/catalog')
@login_required
def api_plugin_catalog():
now = time.time()
if now - _PLUGIN_CATALOG['ts'] > 86400:
try:
r = requests.get('https://plugins.traefik.io/api/services/plugins', timeout=8)
r.raise_for_status()
items = r.json()
catalog = {}
for item in items if isinstance(items, list) else []:
mod = str(item.get('import') or '').strip().lower()
ver = str(item.get('latestVersion') or '').strip()
if mod and ver:
catalog[mod] = ver
if catalog:
_PLUGIN_CATALOG['map'] = catalog
_PLUGIN_CATALOG['ts'] = now
else:
_PLUGIN_CATALOG['ts'] = now - 86400 + 900
except Exception:
_PLUGIN_CATALOG['ts'] = now - 86400 + 900
return jsonify({'plugins': _PLUGIN_CATALOG['map']})
@app.route('/api/plugins/install', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_plugins_install():
data = request.get_json(silent=True) or {}
static_yaml = (data.get('static_yaml') or '').strip()
middleware_yaml = (data.get('middleware_yaml') or '').strip()
if not static_yaml:
return jsonify({'ok': False, 'error': gettext('Paste the static config snippet')}), 400
try:
_ys = SafeYAML(typ='safe')
parsed_static = _ys.load(static_yaml) or {}
except Exception as e:
return jsonify({'ok': False, 'error': gettext('Invalid static config YAML: %(error)s', error=e)}), 400
plugins_block = None
if isinstance(parsed_static.get('experimental'), dict):
plugins_block = parsed_static['experimental'].get('plugins')
if not plugins_block and 'plugins' in parsed_static:
plugins_block = parsed_static['plugins']
if not plugins_block or not isinstance(plugins_block, dict):
return jsonify({'ok': False, 'error': gettext('Could not find plugins block - paste the experimental.plugins YAML from the Traefik plugin page')}), 400
agent = None
static_path = None
server = str(data.get('server') or '').strip()
if server:
agent = _agent_by_id(server)
if not agent:
return jsonify({'ok': False, 'error': gettext('Agent not found')}), 404
try:
resp = _agent_request(agent, 'GET', '/api/static')
except requests.exceptions.RequestException as e:
return jsonify({'ok': False, 'error': gettext('Cannot reach agent: %(error)s', error=e)}), 502
if resp.status_code != 200:
return jsonify({'ok': False, 'error': gettext('Static config not available on this agent')}), 404
source_raw = (resp.json() or {}).get('content', '')
else:
static_path = _get_static_config_path()
if not static_path or not os.path.exists(static_path):
return jsonify({'ok': False, 'error': gettext('Static config not found')}), 404
with open(static_path, 'r') as f:
source_raw = f.read()
try:
_ry = YAML()
_ry.preserve_quotes = True
config = _ry.load(StringIO(source_raw)) or {}
if 'experimental' not in config:
config['experimental'] = {}
if 'plugins' not in config['experimental']:
config['experimental']['plugins'] = {}
for plugin_name, plugin_data in plugins_block.items():
config['experimental']['plugins'][plugin_name] = {
'moduleName': plugin_data.get('moduleName', ''),
'version': plugin_data.get('version', ''),
}
stream = StringIO()
_ry.dump(config, stream)
if agent:
wresp = _agent_request(agent, 'POST', '/api/static', json={'content': stream.getvalue()})
if wresp.status_code != 200:
return jsonify({'ok': False, 'error': gettext('Agent rejected the static config write')}), 502
else:
create_backup(static_path)
with open(static_path, 'w') as f:
f.write(stream.getvalue())
except requests.exceptions.RequestException as e:
return jsonify({'ok': False, 'error': gettext('Cannot reach agent: %(error)s', error=e)}), 502
except Exception as e:
logger.exception("Failed to save plugin to static config")
return jsonify({'ok': False, 'error': str(e)}), 500
warning = None
mw_written = None
if middleware_yaml and not agent and not ACTIVE_CONFIG_DIR:
warning = gettext('Plugin saved, but the middleware was not written - no config directory is configured, so there is no file to write it to')
if middleware_yaml and (agent or ACTIVE_CONFIG_DIR):
if '{{' in middleware_yaml:
return jsonify({'ok': False, 'error': gettext('The middleware snippet contains template placeholders (%(example)s) that must be replaced with real values before saving. Edit the middleware in the editor and replace all %(marker)s placeholders.', example='{{ ... }}', marker='{{ }}')}), 400
try:
_ym = SafeYAML(typ='safe')
parsed_mw = _ym.load(middleware_yaml) or {}
if isinstance(parsed_mw.get('http'), dict):
middlewares = parsed_mw['http'].get('middlewares') or {}
elif 'middlewares' in parsed_mw:
middlewares = parsed_mw['middlewares']
else:
middlewares = {}
if middlewares and isinstance(middlewares, dict):
mw_choice = str(data.get('middleware_file') or '').strip()
existing = {}
if agent:
mw_name = os.path.basename(mw_choice) or 'plugin-middlewares.yml'
if '..' in mw_name:
raise ValueError(_i18n.Message('invalid middleware file name',
lambda: gettext('invalid middleware file name')))
if not mw_name.endswith(('.yml', '.yaml')):
mw_name += '.yml'
mw_label = mw_name
try:
cresp = _agent_request(agent, 'GET', '/api/configs')
for fobj in (cresp.json() or {}).get('files', []):
if fobj.get('name') == mw_name:
existing = yaml.load(StringIO(fobj.get('content', ''))) or {}
break
except Exception:
existing = {}
else:
if mw_choice:
mw_file = _resolve_config_path(mw_choice)
if not mw_file:
raise ValueError(_i18n.Message(
f'middleware file not allowed: {mw_choice!r}',
lambda: gettext('middleware file not allowed: %(file)s', file=repr(mw_choice))))
else:
mw_file = os.path.join(ACTIVE_CONFIG_DIR, 'plugin-middlewares.yml')
mw_label = os.path.basename(mw_file)
if os.path.exists(mw_file):
with open(mw_file, 'r') as f:
existing = yaml.load(f) or {}
if 'http' not in existing:
existing['http'] = {}
if 'middlewares' not in existing['http']:
existing['http']['middlewares'] = {}
existing['http']['middlewares'].update(middlewares)
stream = StringIO()
yaml.dump(existing, stream)
if agent:
mresp = _agent_request(agent, 'POST', '/api/configs', json={'name': mw_name, 'content': stream.getvalue()})
if mresp.status_code != 200:
raise RuntimeError(_i18n.Message('agent rejected the middleware file write',
lambda: gettext('agent rejected the middleware file write')))
else:
if os.path.exists(mw_file):
create_backup(mw_file)
with open(mw_file, 'w') as f:
f.write(stream.getvalue())
mw_written = mw_label
except Exception as e:
logger.exception("Failed to save middleware")
warning = gettext('Plugin saved but middleware could not be written: %(error)s',
error=_i18n.shown_error(e))
plugin_names = list(plugins_block.keys())
add_notification('success', _i18n.lazy_gettext('Plugin installed: %(names)s', names=', '.join(plugin_names)))
result = {'ok': True, 'plugins': plugin_names}
if mw_written:
result['middleware_file'] = mw_written
if warning:
result['warning'] = warning
return jsonify(result)
def _acme_certs_from_paths(acme_paths):
import json as _json
certs = []
errors = []
found_any = False
for configured in acme_paths:
acme_path = _readable_config_path(configured)
if not (acme_path and os.path.exists(acme_path)):
errors.append(f'acme.json not found at {configured}.')
continue
found_any = True
try:
with open(acme_path, 'r') as f:
raw = f.read().strip()
acme_data = _json.loads(raw) if raw else {}
for resolver_name, resolver_data in acme_data.items():
if not isinstance(resolver_data, dict):
continue
for c in (resolver_data.get('Certificates') or resolver_data.get('certificates') or []):
domain = c.get('domain', {})
not_after = _parse_cert_expiry(c.get('certificate', ''))
certs.append({'resolver': resolver_name, 'main': domain.get('main', ''),
'sans': domain.get('sans', []) or [], 'not_after': not_after,
'source': os.path.basename(acme_path)})
except PermissionError:
errors.append(f'Permission denied reading {acme_path}. Run Traefik Manager as the user that owns it, do not chmod it.')
except Exception as e:
logger.exception("Error reading acme.json")
errors.append(f'{os.path.basename(acme_path)}: {e}')
return certs, errors, found_any
def _host_cert_resolvers():
path = _get_static_config_path()
if not path or not os.path.exists(path):
return None
try:
with open(path, 'r') as f:
data = _yaml_safe.load(f) or {}
except Exception:
logger.debug("Failed to read certificatesResolvers from static config", exc_info=True)
return None
resolvers = data.get('certificatesResolvers')
if not isinstance(resolvers, dict) or not resolvers:
return None
return [str(k).strip() for k in resolvers if str(k).strip()]
def _agent_cert_resolvers_or_none(agent):
try:
resp = _agent_request(agent, 'GET', '/api/static')
if resp.status_code != 200:
return None
data = _yaml_safe.load((resp.json() or {}).get('content', '')) or {}
except Exception:
logger.debug("Failed to read agent certificatesResolvers", exc_info=True)
return None
resolvers = data.get('certificatesResolvers')
if not isinstance(resolvers, dict) or not resolvers:
return None
return [str(k).strip() for k in resolvers if str(k).strip()]
@app.route('/api/certs/usage')
@login_required
def api_certs_usage():
server = str(request.args.get('server', '')).strip()
exclude = [str(i).strip() for i in request.args.getlist('exclude') if str(i).strip()][:500]
if server:
agent = _agent_by_id(server)
if not agent:
return jsonify({'error': gettext('Unknown server')}), 404
try:
resp = _agent_request(agent, 'GET', '/api/traefik/certs')
certs = (resp.json() or {}).get('certs') or [] if resp.ok else []
except Exception:
certs = []
payload = _agent_routes_payload(agent, server)
apps = payload.get('apps') or []
configs = list(_agent_load_configs(agent).values())
resolvers = _agent_cert_resolvers_or_none(agent)
ok = not payload.get('configErrors')
result = _cert_usage.analyze(certs, apps, configs, resolvers, routers_ok=ok, configs_ok=ok,
exclude_ids=exclude)
return jsonify(result)
certs, _errors, _found = _acme_certs_from_paths(_settings.get_acme_json_paths())
certs.extend(_certs_from_tls_configs())
complete = []
apps, _mws = _build_all_apps(include_external=True, include_internal=True, complete=complete)
configs = [_cfg._load_config_display(p) for p in env.CONFIG_PATHS]
ok = not complete
result = _cert_usage.analyze(certs, apps, configs, _host_cert_resolvers(),
routers_ok=ok, configs_ok=not _get_config_parse_errors(),
exclude_ids=exclude)
return jsonify(result)
def _host_cert_manage_state():
paths = _settings.get_acme_json_paths()
resolved = [p for p in (_readable_config_path(x) for x in paths) if p and os.path.isfile(p)]
writable = bool(resolved) and all(_acme.writable(p) for p in resolved)
method = _get_restart_method()
restart = method in ('proxy', 'socket', 'poison-pill')
if not resolved:
_acme_refused = _settings.refused_path('acme')
reason = (gettext('acme.json path refused: %(refused)s', refused=_i18n.shown(_acme_refused)) if _acme_refused
else gettext('acme.json is not mounted'))
elif not writable:
reason = gettext('acme.json is mounted read only')
elif not restart:
reason = gettext('no restart method is configured, and Traefik only reads acme.json at startup')
else:
reason = ''
return {'available': writable and restart, 'writable': writable,
'restart_method': method if restart else '', 'reason': reason, 'paths': resolved}
@app.route('/api/certs/manage')
@login_required
def api_certs_manage():
server = str(request.args.get('server', '')).strip()
if not server:
return jsonify(_host_cert_manage_state())
agent = _agent_by_id(server)
if not agent:
return jsonify({'error': gettext('Unknown server')}), 404
try:
resp = _agent_request(agent, 'GET', '/api/traefik/certs/status')
if resp.status_code != 200:
return jsonify({'available': False, 'writable': False, 'restart_method': '',
'reason': gettext('this agent is too old to manage certificates'), 'paths': []})
state = _agent_err.localize(resp.json() or {})
except Exception as e:
return jsonify({'available': False, 'writable': False, 'restart_method': '',
'reason': str(e), 'paths': []})
state['available'] = bool(state.get('available'))
return jsonify(state)
@app.route('/api/certs/delete', methods=['POST'])
@csrf_protect
@login_required
def api_certs_delete():
data = request.get_json(silent=True) or {}
server = str(data.get('server', '')).strip()
wanted = [(str(c.get('resolver', '')), str(c.get('main', '')))
for c in (data.get('certs') or []) if isinstance(c, dict) and c.get('main')]
if not wanted:
return jsonify({'error': gettext('Nothing was selected')}), 400
if server:
agent = _agent_by_id(server)
if not agent:
return jsonify({'error': gettext('Unknown server')}), 404
try:
resp = _agent_request(agent, 'POST', '/api/traefik/certs/delete',
json={'certs': [{'resolver': r, 'main': m} for r, m in wanted]})
return jsonify(_agent_err.localize(resp.json() or {})), resp.status_code
except Exception as e:
return jsonify({'error': str(e)}), 502
state = _host_cert_manage_state()
if not state['available']:
return jsonify({'error': state['reason'] or gettext('Certificates cannot be edited here')}), 403
try:
removed, saved = _acme.remove_many(state['paths'], wanted)
except _acme.AcmeStorePartial as e:
ok, err = trigger_traefik_restart()
logger.error(f"Certificate removal stopped partway: {e}")
add_notification('error', _i18n.lazy_gettext('Certificate removal stopped partway: %(error)s',
error=_error_param(e)), category='traefik')
return jsonify({'error': e.shown(), 'removed': e.removed, 'partial': True,
'backup': os.path.basename(e.backup or ''),
'restarted': ok, 'restart_error': '' if ok else err}), 500
except _acme.AcmeStoreError as e:
return jsonify({'error': e.shown()}), e.status
except OSError as e:
logger.exception("acme.json write failed")
return jsonify({'error': gettext('Could not write acme.json: %(error)s', error=e)}), 500
if not removed:
return jsonify({'error': gettext('No matching certificate was found')}), 404
ok, err = trigger_traefik_restart()
logger.info(f"Removed {removed} certificate(s) from acme.json, backup at {saved}")
add_notification('warning', _i18n.lazy_ngettext('%(num)d certificate removed from acme.json',
'%(num)d certificates removed from acme.json', removed),
category='traefik')
return jsonify({'ok': True, 'removed': removed, 'backup': os.path.basename(saved or ''),
'restarted': ok, 'restart_error': '' if ok else err})
@app.route('/api/traefik/certs')
@login_required
def api_certs():
certs = []
errors = []
acme_paths = _settings.get_acme_json_paths()
certs, errors, found_any = _acme_certs_from_paths(acme_paths)
if not acme_paths or not found_any:
errors.append('Set ACME_JSON_PATH env var or configure the path in Settings. '
'Several files can be given comma-separated, or point it at a directory.')
certs.extend(_certs_from_tls_configs())
if not certs and errors:
return jsonify({'certs': [], 'error': ' | '.join(errors)})
return jsonify({'certs': certs})
@app.route('/api/traefik/logs')
@login_required
def api_logs():
try:
lines_req = int(request.args.get('lines', 100))
except (TypeError, ValueError):
return jsonify({'error': gettext('Invalid lines parameter')}), 400
if lines_req < 1:
return jsonify({'error': gettext('Invalid lines parameter')}), 400
lines_req = min(lines_req, 1000)
log_path = _readable_config_path(_get_access_log_path())
if not log_path or not os.path.exists(log_path):
refused = _settings.refused_path('log')
if refused:
return jsonify({'error': gettext('Access log path refused: %(refused)s', refused=_i18n.shown(refused)), 'lines': []})
return jsonify({'error': gettext('Access log not found. Set ACCESS_LOG_PATH env var or configure the path in Settings.'), 'lines': []})
try:
lines = []
buf_size = 8192
with open(log_path, 'rb') as f:
f.seek(0, 2)
remaining = f.tell()
partial = b''
while remaining > 0 and sum(1 for ln in lines if ln.strip()) < lines_req:
chunk = min(buf_size, remaining)
remaining -= chunk
f.seek(remaining)
data = f.read(chunk) + partial
split = data.split(b'\n')
partial = split[0]
lines = split[1:] + lines
if partial:
lines = [partial] + lines
kept = [ln for ln in lines if ln.strip()]
result = [ln.decode('utf-8', errors='replace').rstrip() for ln in kept[-lines_req:]]
return jsonify({'lines': result})
except Exception as e:
return jsonify({'error': str(e), 'lines': []})
def list_backups():
ensure_backup_dir()
static_path = _get_static_config_path()
static_base = os.path.basename(static_path) if static_path else None
acme_paths = _host_cert_manage_state()['paths']
acme_bases = (set(_acme.backup_keys(acme_paths).values())
| {os.path.basename(p) for p in acme_paths}
| {os.path.basename(p) for p in _settings.get_acme_json_paths()})
_name_re = re.compile(r'^(.+)\.(\d{8}_\d{6})\.bak$')
backups = []
for f in os.listdir(BACKUP_DIR):
if f.endswith('.bak'):
path = os.path.join(BACKUP_DIR, f)
st = os.stat(path)
m = _name_re.match(f)
orig = m.group(1) if m else ''
ts_str = m.group(2) if m else ''
if orig in acme_bases:
kind = 'certs'
elif static_base and orig == static_base:
kind = 'static'
else:
kind = 'routes'
backups.append({
'name': f,
'size': st.st_size,
'modified': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(st.st_mtime)),
'sort_key': ts_str or str(st.st_mtime),
'kind': kind,
})
backups.sort(key=lambda b: b['sort_key'], reverse=True)
for b in backups:
del b['sort_key']
return backups
_BACKUP_RE = re.compile(r'^[a-zA-Z0-9._ -]+\.(yml|yaml|json)\.\d{8}_\d{6}\.bak$')
def _validated_backup_path(filename: str) -> str:
if not _BACKUP_RE.match(filename):
logger.warning(f"Invalid backup filename rejected: {filename!r}")
abort(400)
path = os.path.realpath(os.path.join(BACKUP_DIR, filename))
if not path.startswith(os.path.realpath(BACKUP_DIR)):
logger.warning(f"Path traversal attempt blocked: {filename!r}")
abort(400)
return path
def _git_req_agent():
agent_id = (request.args.get('agent_id') or '').strip()
if not agent_id:
return None, False
return _agent_by_id(agent_id), True
@app.route('/api/backup/git/status')
@login_required
def api_git_backup_status():
agent, wanted = _git_req_agent()
if wanted and not agent:
return jsonify({'error': gettext('Agent not found')}), 404
s = load_settings()
configured = bool(s.get('git_backup_repo', '').strip())
repo_dir = _git_agent_repo_dir(agent['id']) if agent else _git_repo_dir()
result = {'enabled': bool(s.get('git_backup_enabled')), 'configured': configured, 'last_sha': None, 'last_push': None}
if agent:
result['branch'] = _agent_git_branch(agent)
if configured and os.path.exists(os.path.join(repo_dir, '.git')):
out, _, rc = _git_run(['log', '-1', '--format=%H|%ci|%s'], cwd=repo_dir)
if rc == 0 and '|' in out:
parts = out.split('|', 2)
result['last_sha'] = parts[0][:8]
result['last_push'] = parts[1].strip() if len(parts) > 1 else None
return jsonify(result)
@app.route('/api/backup/git/push', methods=['POST'])
@csrf_protect
@login_required
def api_git_backup_push():
agent, wanted = _git_req_agent()
if wanted and not agent:
return jsonify({'error': gettext('Agent not found')}), 404
data = request.get_json(silent=True) or {}
message = str(data.get('message', '')).strip()
if agent:
ok, err = _git_push_agent_configs(agent, 'manual', custom_message=message or None)
else:
ok, err = _git_push_configs('manual', custom_message=message or None)
if ok:
add_notification('success', _i18n.lazy_gettext('Git backup pushed (%(agent)s)', agent=agent['name'])
if agent else _i18n.lazy_gettext('Git backup pushed'), category='backup')
return jsonify({'ok': True})
add_notification('error', _i18n.lazy_gettext('Git push failed: %(error)s', error=err), category='backup')
return jsonify({'ok': False, 'error': _i18n.shown(err)}), 400
def _same_git_remote(a: str, b: str) -> bool:
a, b = (a or '').strip().rstrip('/'), (b or '').strip().rstrip('/')
return bool(a) and bool(b) and a.lower() == b.lower()
@app.route('/api/backup/git/test', methods=['POST'])
@csrf_protect
@login_required
def api_git_backup_test():
body = request.get_json(silent=True) or {}
s = load_settings()
repo_url = (body.get('repo_url') or s.get('git_backup_repo', '')).strip()
username = (body.get('username') or s.get('git_backup_username', '')).strip()
token = (body.get('token') or '').strip()
if not token and _same_git_remote(repo_url, s.get('git_backup_repo', '')):
token = str(s.get('git_backup_token', '')).strip()
if not repo_url:
return jsonify({'ok': False, 'error': gettext('No repository URL configured')}), 400
if not _valid_git_url(repo_url):
return jsonify({'ok': False, 'error': gettext('Unsupported URL - use https://, http://, ssh:// or git://')}), 400
if not _ssrf_ok(repo_url):
return jsonify({'ok': False, 'error': gettext('Target address not allowed')}), 400
creds = {'username': username, 'token': token} if token else None
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
_, err, rc = _git_run(['ls-remote', '--quiet', '--', repo_url], cwd=tmpdir, credentials=creds,
extra_config=['http.followRedirects=false'])
if rc == 0:
return jsonify({'ok': True})
safe_err = err.replace(token, '***') if token else err
return jsonify({'ok': False, 'error': safe_err or gettext('Could not reach repository')}), 400
@app.route('/api/backup/git/commits')
@login_required
def api_git_backup_commits():
agent, wanted = _git_req_agent()
if wanted and not agent:
return jsonify([])
repo_dir = _git_agent_repo_dir(agent['id']) if agent else _git_repo_dir()
if not os.path.exists(os.path.join(repo_dir, '.git')):
return jsonify([])
out, _, rc = _git_run(['log', '--format=%H|%ci|%s', '-50'], cwd=repo_dir)
if rc != 0:
return jsonify([])
commits = []
for line in out.splitlines():
parts = line.split('|', 2)
if len(parts) == 3:
commits.append({'sha': parts[0], 'sha_short': parts[0][:8], 'timestamp': parts[1].strip(), 'message': parts[2].strip()})
return jsonify(commits)
@app.route('/api/backup/git/commit/<sha>/diff')
@login_required
def api_git_backup_diff(sha):
if not re.match(r'^[0-9a-f]{7,40}$', sha):
abort(400)
agent, wanted = _git_req_agent()
if wanted and not agent:
return jsonify({'stat': '', 'files': []})
repo_dir = _git_agent_repo_dir(agent['id']) if agent else _git_repo_dir()
if not os.path.exists(os.path.join(repo_dir, '.git')):
return jsonify({'stat': '', 'files': []})
try:
stat, _, _ = _git_run(['show', '--stat', '--format=', sha], cwd=repo_dir)
changed, _, rc = _git_run(['diff-tree', '--no-commit-id', '-r', '--name-status', sha], cwd=repo_dir)
files = []
for line in changed.splitlines():
line = line.strip()
if not line:
continue
parts = line.split('\t', 1)
if len(parts) != 2:
continue
status, filename = parts[0].strip(), parts[1].strip()
new_content, _, new_rc = _git_run(['show', f'{sha}:{filename}'], cwd=repo_dir)
old_content, _, old_rc = _git_run(['show', f'{sha}^:{filename}'], cwd=repo_dir)
files.append({
'filename': filename,
'status': status,
'old': old_content if old_rc == 0 else '',
'new': new_content if new_rc == 0 else '',
})
return jsonify({'stat': stat, 'files': files})
except Exception as e:
logger.exception("Git diff error")
return jsonify({'error': str(e)}), 500
def _write_restored(path, content):
tmp = f"{path}.tmp.{os.getpid()}.{threading.get_ident()}"
try:
with open(tmp, 'w') as f:
f.write(content)
_cfg._replace_or_copy(tmp, path)
finally:
if os.path.exists(tmp):
os.unlink(tmp)
@app.route('/api/backup/git/restore/<sha>', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_git_backup_restore(sha):
if not re.match(r'^[0-9a-f]{7,40}$', sha):
abort(400)
agent, wanted = _git_req_agent()
if wanted and not agent:
return jsonify({'error': gettext('Agent not found')}), 404
repo_dir = _git_agent_repo_dir(agent['id']) if agent else _git_repo_dir()
if not os.path.exists(os.path.join(repo_dir, '.git')):
return jsonify({'error': gettext('Git repo not initialized')}), 400
try:
if agent:
out, _, rc = _git_run(['ls-tree', '-r', '--name-only', sha, 'dynamic/'], cwd=repo_dir)
if rc != 0:
return jsonify({'error': gettext('Commit not found')}), 404
restored = 0
for fpath in out.splitlines():
fpath = fpath.strip()
if not fpath:
continue
content, _, src = _git_run(['show', f'{sha}:{fpath}'], cwd=repo_dir)
if src == 0:
resp = _agent_request(agent, 'POST', '/api/configs', json={'name': os.path.basename(fpath), 'content': content})
resp.raise_for_status()
restored += 1
add_notification('warning', _i18n.lazy_ngettext('Restored %(agent)s from git commit %(sha)s (%(num)d file)',
'Restored %(agent)s from git commit %(sha)s (%(num)d files)',
restored, agent=agent['name'], sha=sha[:8]),
category='backup')
return jsonify({'ok': True})
for p in env.CONFIG_PATHS:
create_backup(p)
sp = _get_static_config_path()
if sp:
create_backup(sp)
keys = _back.config_keys()
bases = [os.path.basename(p) for p in env.CONFIG_PATHS]
for p in env.CONFIG_PATHS:
base = os.path.basename(p)
candidates = [f'dynamic/{keys[p]}']
if keys[p] == base and bases.count(base) == 1:
candidates.append(base)
content = _git_show_first(repo_dir, sha, candidates)
if content:
_write_restored(p, content)
if sp:
base = os.path.basename(sp)
content = _git_show_first(repo_dir, sha, [f'static/{base}', base])
if content:
_write_restored(sp, content)
add_notification('warning', _i18n.lazy_gettext('Restored from git commit %(sha)s', sha=sha[:8]),
category='backup')
return jsonify({'ok': True})
except Exception as e:
logger.exception("Git restore error")
add_notification('error', _i18n.lazy_gettext('Git restore failed: %(error)s', error=_error_param(e)),
category='backup')
return jsonify({'error': str(e)}), 500
@app.route('/api/backup/git/repo', methods=['DELETE'])
@csrf_protect
@login_required
def api_git_backup_reset():
agent, wanted = _git_req_agent()
if wanted and not agent:
return jsonify({'error': gettext('Agent not found')}), 404
repo_dir = _git_agent_repo_dir(agent['id']) if agent else _git_repo_dir()
try:
if os.path.exists(repo_dir):
shutil.rmtree(repo_dir)
logger.info("Git repo directory reset by user")
add_notification('warning', _i18n.lazy_gettext('Git repository reset - re-initialize by pushing again'),
category='backup')
return jsonify({'ok': True})
except Exception as e:
logger.exception("Git repo reset error")
return jsonify({'error': str(e)}), 500
@app.route('/api/notifications')
@login_required
def api_notifications():
return jsonify(_noti.shown_entries(get_notifications()))
@app.route('/api/notifications/log', methods=['POST'])
@csrf_protect
@login_required
def api_notifications_log():
data = request.get_json(silent=True) or {}
msg = str(data.get('message', ''))[:300].strip()
type_ = str(data.get('type', 'info')).lower()
if type_ not in ('info', 'success', 'warning', 'error'):
type_ = 'info'
if not msg:
return jsonify({'ok': False}), 400
category = str(data.get('category', '')).strip().lower()
if category not in _settings.CHANNEL_CATEGORIES:
category = 'config'
stored = add_notification(type_, msg, category=category, webhook=False)
return jsonify({'ok': True, 'stored': stored})
@app.route('/api/notifications/delete', methods=['POST'])
@csrf_protect
@login_required
def api_notifications_delete():
data = request.get_json(silent=True) or {}
if 'id' in data:
if not _noti.delete_notification_by_id(data.get('id')):
return jsonify({'ok': False, 'message': gettext('Notification not found')}), 404
return jsonify({'ok': True})
ts = data.get('ts', '')
if not ts:
return jsonify({'ok': False, 'message': gettext('Missing id or ts')}), 400
delete_notification(ts)
return jsonify({'ok': True})
@app.route('/api/notifications/read', methods=['POST'])
@csrf_protect
@login_required
def api_notifications_read():
data = request.get_json(silent=True) or {}
if data.get('all'):
marker = _noti.highest_id()
else:
try:
marker = int(data.get('id'))
except (TypeError, ValueError):
return jsonify({'ok': False, 'message': gettext('Missing id or all')}), 400
update_settings(notifications_read_until=max(0, marker))
return jsonify({'ok': True, 'read_until': max(0, marker)})
@app.route('/api/notifications/state')
@login_required
def api_notifications_state():
entries = _noti.get_notifications()
marker = int(load_settings().get('notifications_read_until', 0) or 0)
unread = sum(1 for e in entries if int(e.get('id', 0) or 0) > marker)
return jsonify({'read_until': marker, 'count': len(entries), 'unread': unread})
@app.route('/api/notifications/clear', methods=['POST'])
@csrf_protect
@login_required
def api_notifications_clear():
clear_notifications()
return jsonify({'ok': True})
@app.route('/api/notifications/add', methods=['POST'])
@csrf_protect
@login_required
def api_notifications_add():
data = request.get_json(silent=True) or {}
type_ = data.get('type', 'info')
msg = (data.get('message') or '').strip()
if not msg:
return jsonify({'ok': False, 'error': gettext('message required')}), 400
category = str(data.get('category', '')).strip().lower()
if category not in _settings.CHANNEL_CATEGORIES:
category = 'config'
add_notification(type_, msg, category=category)
return jsonify({'ok': True})
@app.route('/api/notifications/update', methods=['POST'])
@csrf_protect
@login_required
def api_notifications_update():
data = request.get_json(silent=True) or {}
version = data.get('version', '')
product = 'Traefik Manager' if data.get('product') == 'manager' else 'Traefik'
if version:
add_notification('info', _i18n.lazy_gettext('%(product)s v%(version)s is available - update now',
product=product, version=version), category='update')
return jsonify({'ok': True})
_CHANNEL_SECRETS = ('token', 'token2', 'password')
_CHANNEL_SECRET_URL_KINDS = ('discord', 'slack', 'ntfy', 'generic')
def _mask_channel_url(kind: str, url: str) -> str:
if not url or kind not in _CHANNEL_SECRET_URL_KINDS:
return url
try:
parts = urlparse(url)
if not parts.scheme or not parts.netloc:
return '***'
return f'{parts.scheme}://{parts.netloc}/***'
except Exception:
return '***'
_QUIET_HOURS_RE = re.compile(r'^(?:[01]\d|2[0-3]):[0-5]\d-(?:[01]\d|2[0-3]):[0-5]\d$')
def _redact_channel(c: dict) -> dict:
out = dict(c)
for field in _CHANNEL_SECRETS:
out[field] = '***' if out.get(field) else ''
out['url'] = _mask_channel_url(out.get('kind', ''), out.get('url', ''))
return out
def _blank_channel() -> dict:
return {
'id': _settings._new_channel_id(),
'name': '',
'kind': '',
'enabled': True,
'url': '',
'token': '',
'token2': '',
'username': '',
'password': '',
'categories': list(_settings.CHANNEL_CATEGORIES),
'min_severity': 'info',
'digest': 'immediate',
'quiet_hours': '',
'break_through': False,
}
def _apply_channel_fields(data, base, require_kind):
ch = dict(base)
if require_kind or 'kind' in data:
kind = str(data.get('kind', '')).strip().lower()
if kind not in _settings.CHANNEL_KINDS:
return None, gettext('kind must be one of: %(channel_kinds)s', channel_kinds=', '.join(_settings.CHANNEL_KINDS))
ch['kind'] = kind
if 'categories' in data:
raw = data.get('categories')
if not isinstance(raw, list):
return None, gettext('categories must be a list')
unknown = [str(c) for c in raw if c not in _settings.CHANNEL_CATEGORIES]
if unknown:
return None, gettext('unknown categories: %(unknown)s', unknown=', '.join(unknown))
ch['categories'] = list(raw)
if 'min_severity' in data:
sev = str(data.get('min_severity', '')).strip().lower()
if sev not in _settings.CHANNEL_SEVERITIES:
return None, gettext('min_severity must be one of: %(channel_severities)s', channel_severities=', '.join(_settings.CHANNEL_SEVERITIES))
ch['min_severity'] = sev
if 'digest' in data:
digest = str(data.get('digest', '')).strip().lower()
if digest not in _settings.CHANNEL_DIGESTS:
return None, gettext('digest must be one of: %(channel_digests)s', channel_digests=', '.join(_settings.CHANNEL_DIGESTS))
ch['digest'] = digest
if 'quiet_hours' in data:
window = str(data.get('quiet_hours', '')).strip()
if window and not _QUIET_HOURS_RE.match(window):
return None, gettext('quiet_hours must be HH:MM-HH:MM, for example 23:00-07:00')
ch['quiet_hours'] = window
if 'name' in data:
ch['name'] = str(data.get('name', '')).strip()[:60]
if 'url' in data:
incoming_url = str(data.get('url', '')).strip()[:500]
if '***' not in incoming_url:
ch['url'] = incoming_url
if 'username' in data:
ch['username'] = str(data.get('username', '')).strip()[:100]
if 'enabled' in data:
ch['enabled'] = bool(data.get('enabled'))
if 'break_through' in data:
ch['break_through'] = bool(data.get('break_through'))
for field in _CHANNEL_SECRETS:
if field in data and str(data[field]) != '***':
ch[field] = str(data[field])
return ch, ''
def _save_channels(settings, channels):
update_settings(notification_channels=channels)
def _stored_channel(channel_id, fallback):
channels = load_settings().get('notification_channels', [])
return next((c for c in channels if c.get('id') == channel_id), fallback)
@app.route('/api/notifications/channels', methods=['GET'])
@login_required
def api_notification_channels_list():
channels = load_settings().get('notification_channels', [])
return jsonify({'channels': [_redact_channel(c) for c in channels]})
@app.route('/api/notifications/channels', methods=['POST'])
@csrf_protect
@login_required
def api_notification_channels_create():
data = request.get_json(silent=True) or {}
settings = load_settings()
channel, err = _apply_channel_fields(data, _blank_channel(), True)
if err:
return jsonify({'ok': False, 'error': err}), 400
channel['name'] = channel['name'] or channel['kind'].title()
channels = list(settings.get('notification_channels', []))
channels.append(channel)
_save_channels(settings, channels)
logger.info(f"Notification channel '{channel['name']}' ({channel['kind']}) created by {request.remote_addr}")
return jsonify({'ok': True, 'channel': _redact_channel(_stored_channel(channel['id'], channel))})
@app.route('/api/notifications/channels/<channel_id>', methods=['PUT'])
@csrf_protect
@login_required
def api_notification_channels_update(channel_id):
data = request.get_json(silent=True) or {}
settings = load_settings()
channels = list(settings.get('notification_channels', []))
idx = next((i for i, c in enumerate(channels) if c.get('id') == channel_id), None)
if idx is None:
return jsonify({'ok': False, 'error': gettext('Channel not found')}), 404
channel, err = _apply_channel_fields(data, channels[idx], False)
if err:
return jsonify({'ok': False, 'error': err}), 400
channel['id'] = channel_id
channels[idx] = channel
_save_channels(settings, channels)
return jsonify({'ok': True, 'channel': _redact_channel(_stored_channel(channel_id, channel))})
@app.route('/api/notifications/channels/<channel_id>', methods=['DELETE'])
@csrf_protect
@login_required
def api_notification_channels_delete(channel_id):
settings = load_settings()
current = settings.get('notification_channels', [])
channels = [c for c in current if c.get('id') != channel_id]
if len(channels) == len(current):
return jsonify({'ok': False, 'error': gettext('Channel not found')}), 404
_save_channels(settings, channels)
logger.info(f"Notification channel {channel_id} deleted by {request.remote_addr}")
return jsonify({'ok': True})
@app.route('/api/notifications/channels/<channel_id>/test', methods=['POST'])
@csrf_protect
@login_required
def api_notification_channels_test(channel_id):
channels = load_settings().get('notification_channels', [])
channel = next((c for c in channels if c.get('id') == channel_id), None)
if channel is None:
return jsonify({'ok': False, 'error': gettext('Channel not found')}), 404
missing = _notify_providers.missing_fields(channel)
if missing:
return jsonify({'ok': False, 'error': gettext('Channel is missing %(missing)s', missing=', '.join(missing))}), 400
ts = time.strftime("%Y-%m-%d %H:%M:%S")
ok, detail = _notify_providers.send(channel, 'info', 'Traefik Manager',
'Traefik Manager test notification', ts)
if not ok:
return jsonify({'ok': False, 'error': detail[:200]})
return jsonify({'ok': True, 'detail': detail[:200]})
def _tls_opt_sources(server):
agent = _agent_by_id(server) if server else None
if agent:
cfgs = _agent_load_configs(agent)
return [(name, cfg, name) for name, cfg in cfgs.items()], True
out = []
for p in env.CONFIG_PATHS:
short = os.path.basename(p) if (env.MULTI_CONFIG or ACTIVE_CONFIG_DIR) else ''
out.append((short, _load_config_display(p), p))
return out, False
@app.route('/api/tls-options')
@login_required
def api_tls_options_list():
opts = []
sources, _is_agent = _tls_opt_sources(request.args.get('server', ''))
for short, config, p in sources:
for name, data in (config.get('tls') or {}).get('options', {}).items():
data = data or {}
buf = StringIO()
yaml.dump({'tls': {'options': {name: data}}}, buf)
ca = data.get('clientAuth') or {}
opts.append({
'name': name,
'configFile': short,
'configFilePath': p,
'minVersion': data.get('minVersion', ''),
'maxVersion': data.get('maxVersion', ''),
'sniStrict': bool(data.get('sniStrict', False)),
'cipherSuites': data.get('cipherSuites', []),
'curvePreferences': data.get('curvePreferences', []),
'alpnProtocols': data.get('alpnProtocols', []),
'clientAuthType': ca.get('clientAuthType', ''),
'clientAuthCAs': ca.get('caFiles', []),
'yaml': buf.getvalue(),
})
return jsonify(opts)
@app.route('/api/tls-options', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_tls_options_save():
data = request.get_json(silent=True) or {}
name = data.get('name', '').strip()
config_file = data.get('configFile', '').strip()
if not name:
return jsonify({'ok': False, 'message': gettext('Profile name is required')}), 400
server = str(data.get('server') or request.args.get('server', '')).strip()
agent = _agent_by_id(server) if server else None
if agent:
cfgs = _agent_load_configs(agent)
cfg_name = config_file or (next(iter(cfgs), 'dynamic.yml'))
config = cfgs.get(cfg_name, {})
target_path = None
else:
target_path = _resolve_config_path(config_file) or env.CONFIG_PATH
create_backup(target_path)
config = load_config(target_path)
opts = {}
if data.get('minVersion'):
opts['minVersion'] = data['minVersion']
if data.get('maxVersion'):
opts['maxVersion'] = data['maxVersion']
if data.get('sniStrict'):
opts['sniStrict'] = True
ciphers = [c.strip() for c in (data.get('cipherSuites') or []) if c.strip()]
if ciphers:
opts['cipherSuites'] = ciphers
curves = [c.strip() for c in (data.get('curvePreferences') or []) if c.strip()]
if curves:
opts['curvePreferences'] = curves
alpn = [c.strip() for c in (data.get('alpnProtocols') or []) if c.strip()]
if alpn:
opts['alpnProtocols'] = alpn
ca_type = data.get('clientAuthType', '').strip()
ca_cas = [c.strip() for c in (data.get('clientAuthCAs') or []) if c.strip()]
if ca_type and ca_type != 'NoClientCert':
ca_obj = {'clientAuthType': ca_type}
if ca_cas:
ca_obj['caFiles'] = ca_cas
opts['clientAuth'] = ca_obj
original = str(data.get('originalName') or '').strip()
if original and original != name:
(config.get('tls') or {}).get('options', {}).pop(original, None)
_retarget_tls_option(config, original, name)
config.setdefault('tls', {}).setdefault('options', {})[name] = opts
if agent:
_agent_write_config(agent, cfg_name, config)
else:
save_config(_strip_empty_sections(config), target_path)
if original and original != name:
_cascade_across_configs(agent, lambda c: _retarget_tls_option(c, original, name),
already=cfg_name if agent else target_path)
add_notification('success', _i18n.lazy_gettext("TLS profile '%(name)s' saved", name=name))
return jsonify({'ok': True})
@app.route('/api/tls-options/<name>', methods=['DELETE'])
@csrf_protect
@login_required
@_config_edit
def api_tls_options_delete(name):
config_file = request.args.get('configFile', '').strip()
server = request.args.get('server', '').strip()
agent = _agent_by_id(server) if server else None
if agent:
cfgs = _agent_load_configs(agent)
cfg_name = config_file or (next(iter(cfgs), 'dynamic.yml'))
config = cfgs.get(cfg_name, {})
target_path = None
else:
target_path = _resolve_config_path(config_file) or env.CONFIG_PATH
config = load_config(target_path)
tls_opts = (config.get('tls') or {}).get('options', {})
if name not in tls_opts:
return jsonify({'ok': False, 'message': gettext('Profile not found')}), 404
_tls_all = (list(cfgs.values()) if agent else [load_config(_p) for _p in env.CONFIG_PATHS])
_tls_users = _tls_option_routers_using(_tls_all, name)
if _tls_users:
return jsonify({'ok': False,
'message': gettext('%(child)s is still used by %(users)s', child=name, users=_name_list(_tls_users)),
'inUseBy': _tls_users}), 409
del tls_opts[name]
if agent:
_agent_write_config(agent, cfg_name, _strip_empty_sections(config))
else:
create_backup(target_path)
save_config(_strip_empty_sections(config), target_path)
add_notification('success', _i18n.lazy_gettext("TLS profile '%(name)s' deleted", name=name))
return jsonify({'ok': True})
@app.route('/api/backups')
@login_required
def api_backups():
return jsonify(list_backups())
@app.route('/api/restore/<filename>', methods=['POST'])
@limiter.limit("10 per minute")
@csrf_protect
@login_required
@_config_edit
def api_restore(filename):
try:
path = _validated_backup_path(filename)
if not os.path.exists(path):
return jsonify({'error': gettext('Backup not found')}), 404
stamped = re.match(r'^(.+)\.\d{8}_\d{6}\.bak$', filename)
orig = stamped.group(1) if stamped else ''
keys = _back.config_keys()
by_stem = [p for p in env.CONFIG_PATHS if _back.backup_stem(keys[p]) == orig]
by_base = [p for p in env.CONFIG_PATHS if os.path.basename(p) == orig]
if orig and not by_stem and len(by_base) > 1:
return jsonify({'error': gettext('%(original)s matches more than one config file, so this backup cannot be restored safely', original=orig)}), 409
target_path = (by_stem or by_base or [None])[0]
if target_path is None:
static_path = _get_static_config_path()
if static_path and os.path.basename(static_path) == orig:
target_path = static_path
acme_target = None
acme_key = None
if target_path is None and orig:
store_paths = _host_cert_manage_state()['paths']
for store_path, key in _acme.backup_keys(store_paths).items():
if key == orig:
acme_target, acme_key = store_path, key
break
if acme_target is None and sum(1 for p in store_paths if os.path.basename(p) == orig) > 1:
return jsonify({'error': gettext('%(original)s matches more than one certificate store, so this backup cannot be restored safely', original=orig)}), 409
if acme_target:
state = _host_cert_manage_state()
if not state['available']:
return jsonify({'error': state['reason'] or gettext('acme.json cannot be written here')}), 403
with open(path, 'rb') as fh:
body = fh.read()
with _acme.store_lock():
current = _acme.snapshot(acme_target)
_acme.backup(acme_target, current, acme_key)
_acme.write_bytes_in_place(acme_target, body, restore=current)
ok, err = trigger_traefik_restart()
logger.info(f"Restored: {filename} -> {acme_target}")
add_notification('warning', _i18n.lazy_gettext('Certificate store restored: %(file)s', file=filename),
category='backup')
return jsonify({'success': True, 'restarted': ok, 'restart_error': '' if ok else err})
if target_path is None:
return jsonify({'error': gettext('No config file matches %(filename)s', filename=repr(filename))}), 400
with open(path, 'rb') as fh:
restored = fh.read()
create_backup(target_path)
with open(target_path, 'wb') as fh:
fh.write(restored)
logger.info(f"Restored: {filename} → {target_path}")
add_notification('warning', _i18n.lazy_gettext('Backup restored: %(file)s', file=filename), category='backup')
return jsonify({'success': True})
except _acme.AcmeStoreError as e:
return jsonify({'error': e.shown()}), e.status
except Exception as e:
logger.exception("Restore error")
return jsonify({'error': str(e)}), 500
@app.route('/api/backup/create', methods=['POST'])
@csrf_protect
@login_required
def api_backup_create():
try:
created = []
for p in env.CONFIG_PATHS:
dest = create_backup(p)
if dest:
created.append(os.path.basename(dest))
if created:
add_notification('success', _i18n.lazy_ngettext('Backup created (%(num)d file)',
'Backup created (%(num)d files)', len(created)),
category='backup')
return jsonify({'success': True, 'names': created, 'count': len(created)})
return jsonify({'error': gettext('No config files found to backup')}), 400
except Exception as e:
logger.exception("Backup create error")
return jsonify({'error': str(e)}), 500
@app.route('/api/static/backup/create', methods=['POST'])
@csrf_protect
@login_required
def api_static_backup_create():
path = _get_static_config_path()
if not path:
return jsonify({'error': gettext('STATIC_CONFIG_PATH not configured')}), 400
try:
dest = create_backup(path)
if dest:
add_notification('success', _i18n.lazy_gettext('Static config backup created'), category='backup')
return jsonify({'success': True, 'name': os.path.basename(dest)})
return jsonify({'error': gettext('Static config file not found')}), 400
except Exception as e:
logger.exception("Static backup create error")
return jsonify({'error': str(e)}), 500
@app.route('/api/backup/static/create', methods=['POST'])
@csrf_protect
@login_required
def api_backup_static_create_alias():
return api_static_backup_create()
@app.route('/api/backup/delete/<filename>', methods=['POST'])
@csrf_protect
@login_required
def api_backup_delete(filename):
try:
path = _validated_backup_path(filename)
if os.path.exists(path):
os.remove(path)
add_notification('warning', _i18n.lazy_gettext('Backup deleted: %(file)s', file=filename), category='backup')
return jsonify({'success': True})
except Exception as e:
logger.exception("Backup delete error")
return jsonify({'error': str(e)}), 500
def _redact_channels(channels) -> list:
return [_redact_channel(c) for c in (channels or []) if isinstance(c, dict)]
def _merge_channel_secrets(incoming, existing) -> list:
by_id = {str(c.get('id', '')): c for c in (existing or []) if isinstance(c, dict)}
out = []
for item in (incoming or []):
if not isinstance(item, dict):
continue
c = dict(item)
old = by_id.get(str(c.get('id', '')), {})
for field in _CHANNEL_SECRETS:
if str(c.get(field, '')) in ('', '***'):
c[field] = old.get(field, '')
if '***' in str(c.get('url', '')):
c['url'] = old.get('url', '')
out.append(c)
return out
@app.route('/api/settings', methods=['GET'])
@login_required
def api_get_settings():
s = load_settings()
s.pop('password_hash', None)
s.pop('oidc_client_secret', None)
s.pop('webhook_password', None)
s.pop('crowdsec_api_key', None)
s.pop('crowdsec_machine_password', None)
s.pop('traefik_api_password', None)
s.pop('otp_secret', None)
s.pop('agents', None)
s['traefik_api_password_set'] = bool(load_settings().get('traefik_api_password', ''))
s['auth_enabled'] = _auth_enabled()
s['oidc_active'] = _oidc_active()
s['no_auth'] = not _auth_required()
s['auth_external_ack'] = bool(s.get('auth_external_ack'))
s['has_password'] = _has_password_set()
s['auth_env_forced'] = os.environ.get('AUTH_ENABLED', '').strip().lower() in ('false', '0', 'no')
s['oidc_client_secret_set'] = bool(load_settings().get('oidc_client_secret', ''))
s['crowdsec_api_key_set'] = bool(_cs_api_key())
s['crowdsec_machine_password_set'] = bool(_cs_machine_password())
s['crowdsec_enabled'] = bool(_cs_lapi_url() and (_cs_api_key() or _cs_has_machine()))
s['git_backup_token_set'] = bool(s.get('git_backup_token', ''))
s.pop('git_backup_token', None)
s['notification_channels'] = _redact_channels(s.get('notification_channels'))
s['available_languages'] = _i18n.language_options()
return jsonify(s)
@app.route('/api/settings', methods=['POST'])
@csrf_protect
@login_required
def api_save_settings():
try:
data = request.get_json()
domains_raw = data.get('domains', '')
domains = _cfg.parse_domains(domains_raw)
if not domains:
return jsonify({'error': gettext('At least one domain is required')}), 400
cert_resolver = str(data.get('cert_resolver', 'cloudflare')).strip()
traefik_api_url = _safe_api_url(str(data.get('traefik_api_url', 'http://traefik:8080')))
if not traefik_api_url:
return jsonify({'error': gettext('Invalid traefik_api_url - must start with http:// or https://')}), 400
acme_json_path = str(data.get('acme_json_path', '')).strip()
access_log_path = str(data.get('access_log_path', '')).strip()
static_config_path = str(data.get('static_config_path', '')).strip()
_path_errors = {
'static': lambda p: gettext('Static config path: %(problem)s', problem=p),
'log': lambda p: gettext('Access log path: %(problem)s', problem=p),
'acme': lambda p: gettext('acme.json path: %(problem)s', problem=p),
}
for _kind, _value in (('static', static_config_path), ('log', access_log_path),
('acme', acme_json_path)):
_problem = _settings.saved_path_problem(_kind, _value)
if _problem:
return jsonify({'error': _path_errors[_kind](_i18n.shown(_problem))}), 400
webhook_url = str(data.get('webhook_url', '')).strip()
webhook_type = str(data.get('webhook_type', 'discord')).strip()
webhook_username = str(data.get('webhook_username', '')).strip()
webhook_password = str(data.get('webhook_password', ''))
crowdsec_lapi_url = str(data.get('crowdsec_lapi_url', '')).strip()
crowdsec_api_key = str(data.get('crowdsec_api_key', ''))
crowdsec_alert_limit = str(data.get('crowdsec_alert_limit', '')).strip()
if crowdsec_alert_limit:
try:
_lim = int(crowdsec_alert_limit)
if _lim < 0 or _lim > 100000:
return jsonify({'error': gettext('Alert limit must be between 0 and 100000')}), 400
crowdsec_alert_limit = str(_lim)
except ValueError:
return jsonify({'error': gettext('Alert limit must be a whole number')}), 400
crowdsec_machine_id = str(data.get('crowdsec_machine_id', '')).strip()
crowdsec_machine_password = str(data.get('crowdsec_machine_password', ''))
crowdsec_client_cert = str(data.get('crowdsec_client_cert', '')).strip()
crowdsec_client_key = str(data.get('crowdsec_client_key', '')).strip()
crowdsec_ca_cert = str(data.get('crowdsec_ca_cert', '')).strip()
traefik_api_user = str(data.get('traefik_api_user', '')).strip()
traefik_api_password = str(data.get('traefik_api_password', ''))
git_backup_enabled = bool(data['git_backup_enabled']) if 'git_backup_enabled' in data else None
git_backup_repo = str(data['git_backup_repo']).strip() if 'git_backup_repo' in data else None
if git_backup_repo and not _valid_git_url(git_backup_repo):
return jsonify({'error': gettext('Invalid git repository URL - must start with https://, http://, ssh:// or git://')}), 400
git_backup_branch = (str(data['git_backup_branch']).strip() or 'main') if 'git_backup_branch' in data else None
git_backup_username = str(data['git_backup_username']).strip() if 'git_backup_username' in data else None
git_backup_token = str(data.get('git_backup_token', ''))
git_backup_commit_message = (str(data['git_backup_commit_message']).strip() or 'traefik-manager: {action} at {timestamp}') if 'git_backup_commit_message' in data else None
git_backup_auto_push = bool(data['git_backup_auto_push']) if 'git_backup_auto_push' in data else None
backup_keep_count = max(0, int(data['backup_keep_count'])) if str(data.get('backup_keep_count', '')).strip() != '' else None
default_theme = str(data['default_theme']).strip().lower() if 'default_theme' in data else None
existing = load_settings()
notification_channels = None
if 'notification_channels' in data and isinstance(data['notification_channels'], list):
notification_channels = _merge_channel_secrets(data['notification_channels'],
existing.get('notification_channels', []))
if not webhook_password:
webhook_password = existing.get('webhook_password', '')
if not crowdsec_api_key:
crowdsec_api_key = existing.get('crowdsec_api_key', '')
if not crowdsec_machine_password:
crowdsec_machine_password = existing.get('crowdsec_machine_password', '')
if not traefik_api_password:
if (existing.get('traefik_api_password') and traefik_api_user
and not _same_api_origin(traefik_api_url, existing.get('traefik_api_url', ''))):
return jsonify({'error': gettext('Re-enter the Traefik API password when changing the API URL')}), 400
traefik_api_password = existing.get('traefik_api_password', '')
if not git_backup_token:
git_backup_token = existing.get('git_backup_token', '')
save_settings(domains, cert_resolver, traefik_api_url,
auth_enabled=existing['auth_enabled'],
password_hash=existing['password_hash'],
visible_tabs=existing['visible_tabs'],
acme_json_path=acme_json_path,
access_log_path=access_log_path,
static_config_path=static_config_path,
webhook_url=webhook_url,
webhook_type=webhook_type,
webhook_username=webhook_username,
webhook_password=webhook_password,
crowdsec_lapi_url=crowdsec_lapi_url,
crowdsec_api_key=crowdsec_api_key,
crowdsec_alert_limit=crowdsec_alert_limit,
crowdsec_machine_id=crowdsec_machine_id,
crowdsec_machine_password=crowdsec_machine_password,
crowdsec_client_cert=crowdsec_client_cert,
crowdsec_client_key=crowdsec_client_key,
crowdsec_ca_cert=crowdsec_ca_cert,
traefik_api_user=traefik_api_user,
traefik_api_password=traefik_api_password,
git_backup_enabled=git_backup_enabled,
git_backup_repo=git_backup_repo,
git_backup_branch=git_backup_branch,
git_backup_username=git_backup_username,
git_backup_token=git_backup_token,
git_backup_commit_message=git_backup_commit_message,
git_backup_auto_push=git_backup_auto_push,
backup_keep_count=backup_keep_count,
notification_channels=notification_channels,
default_theme=default_theme)
result = load_settings()
for _k in ('password_hash', 'oidc_client_secret', 'crowdsec_api_key',
'crowdsec_machine_password', 'traefik_api_password', 'git_backup_token',
'webhook_password', 'otp_secret', 'agents'):
result.pop(_k, None)
result['notification_channels'] = _redact_channels(result.get('notification_channels'))
return jsonify({'success': True, 'settings': result})
except Exception as e:
logger.exception("Settings save error")
return jsonify({'error': str(e)}), 500
@app.route('/api/settings/webhook-test', methods=['POST'])
@csrf_protect
@login_required
def api_webhook_test():
data = request.get_json(silent=True) or {}
url = str(data.get('url', '')).strip()
wtype = str(data.get('webhook_type', 'discord')).strip()
username = str(data.get('username', '')).strip()
password = str(data.get('password', ''))
if not url or not url.startswith(('http://', 'https://')):
return jsonify({'ok': False, 'error': gettext('Invalid URL')}), 400
if not _ssrf_ok(url):
return jsonify({'ok': False, 'error': gettext('Target address not allowed')}), 400
ts = time.strftime("%Y-%m-%d %H:%M:%S")
try:
_send_webhook(url, wtype, 'info', 'Traefik Manager webhook test', ts, username, password)
return jsonify({'ok': True})
except Exception as e:
return jsonify({'ok': False, 'error': str(e)[:120]})
@app.route('/api/settings/test-connection', methods=['POST'])
@csrf_protect
@login_required
def api_settings_test_connection():
data = request.get_json(silent=True) or {}
raw_url = str(data.get('url', '')).strip()
url = _safe_api_url(raw_url)
if not url:
return jsonify({'ok': False, 'error': gettext('Invalid URL')}), 400
if not _ssrf_ok(url):
return jsonify({'ok': False, 'error': gettext('Target address not allowed')}), 400
u = str(data.get('user', '')).strip()
p = str(data.get('password', '')).strip()
withheld = False
if not p:
stored = load_settings()
if _same_api_origin(url, stored.get('traefik_api_url', '')):
if not u:
u = stored.get('traefik_api_user', '')
p = stored.get('traefik_api_password', '')
else:
withheld = bool(stored.get('traefik_api_password'))
auth = (u, p) if u and p else None
logger.info(f"Connection test to {url!r} by {request.remote_addr}")
try:
resp = requests.get(f"{url}/api/version", timeout=4, auth=auth,
verify=_traefik_verify(), allow_redirects=False)
if resp.status_code == 200:
info = resp.json()
return jsonify({'ok': True, 'version': info.get('Version', '?')})
if resp.status_code in (401, 403):
if withheld:
err = gettext('HTTP %(status_code)s - the saved password is only sent to the saved API URL, '
'enter it to test this one', status_code=resp.status_code)
else:
err = gettext('HTTP %(status_code)s - check the API username and password',
status_code=resp.status_code)
return jsonify({'ok': False, 'error': err})
return jsonify({'ok': False, 'error': gettext('HTTP %(status_code)s from %(url)s/api/version', status_code=resp.status_code, url=url)})
except requests.exceptions.SSLError as e:
return jsonify({'ok': False, 'error': _traefik_api_unreachable_error(e, url)})
except requests.exceptions.Timeout:
return jsonify({'ok': False, 'error': gettext('Connection timed out - the API URL may be unreachable from the container')})
except requests.exceptions.ConnectionError as e:
return jsonify({'ok': False, 'error': _traefik_api_unreachable_error(e, url)})
except Exception as e:
return jsonify({'ok': False, 'error': str(e)[:160]})
@app.route('/api/settings/tabs', methods=['POST'])
@csrf_protect
@login_required
def api_save_tabs():
try:
data = request.get_json() or {}
existing = load_settings()
vt = existing['visible_tabs'].copy()
for t in OPTIONAL_TABS:
if t in data:
vt[t] = bool(data[t])
save_settings(
domains=existing['domains'],
cert_resolver=existing['cert_resolver'],
traefik_api_url=existing['traefik_api_url'],
auth_enabled=existing['auth_enabled'],
password_hash=existing['password_hash'],
visible_tabs=vt,
)
logger.info(f"Tab visibility updated by {request.remote_addr}: {vt}")
return jsonify({'success': True, 'visible_tabs': vt})
except Exception as e:
logger.exception("Tab settings save error")
return jsonify({'error': str(e)}), 500
@app.route('/api/settings/ui', methods=['GET', 'POST'])
@csrf_protect
@login_required
def api_ui_prefs():
existing = load_settings()
if request.method == 'GET':
return jsonify({'ok': True, 'ui_prefs': existing.get('ui_prefs', {})})
data = request.get_json(silent=True) or {}
incoming = data.get('ui_prefs', data)
if not isinstance(incoming, dict):
return jsonify({'ok': False, 'message': gettext('ui_prefs must be an object')}), 400
merged = dict(existing.get('ui_prefs', {}))
merged.update(_settings.sanitize_ui_prefs(incoming))
update_settings(ui_prefs=merged)
return jsonify({'ok': True, 'ui_prefs': merged})
@app.route('/api/settings/language', methods=['POST'])
@csrf_protect
@login_required
def api_save_language():
try:
data = request.get_json(silent=True) or {}
raw = str(data.get('default_language', '')).strip()
tag = _i18n.normalize(raw)
if raw and not tag:
return jsonify({'success': False, 'error': gettext('Unknown language')}), 400
update_settings(default_language=tag or '')
return jsonify({'success': True, 'default_language': tag or ''})
except Exception:
logger.exception("Language save error")
return jsonify({'success': False, 'error': gettext('Save failed')}), 500
@app.route('/api/settings/theme', methods=['POST'])
@csrf_protect
@login_required
def api_save_theme():
try:
data = request.get_json(silent=True) or {}
theme = str(data.get('default_theme', '')).strip().lower()
if theme not in ('dark', 'light', 'system'):
return jsonify({'success': False, 'error': gettext('Invalid theme')}), 400
update_settings(default_theme=theme)
return jsonify({'success': True, 'default_theme': theme})
except Exception:
logger.exception("Theme save error")
return jsonify({'success': False, 'error': gettext('Save failed')}), 500
@app.route('/api/settings/geoip', methods=['POST'])
@csrf_protect
@login_required
def api_save_geoip():
try:
data = request.get_json(silent=True) or {}
existing = load_settings()
enabled = bool(data['geoip_enabled']) if 'geoip_enabled' in data else existing.get('geoip_enabled', False)
db_path = str(data.get('geoip_db_path', existing.get('geoip_db_path', ''))).strip()
update_settings(geoip_enabled=enabled,
geoip_db_path=db_path)
return jsonify({'success': True, 'status': _geoip_status()})
except Exception:
logger.exception("GeoIP settings save error")
return jsonify({'success': False, 'error': gettext('Save failed')}), 500
@app.route('/api/routes/health')
@login_required
def api_routes_health():
agent_id = request.args.get('agent_id', '').strip()
if agent_id and not _agent_by_id(agent_id):
return jsonify({'error': gettext('Agent not found')}), 404
return jsonify(_rh.snapshot(agent_id or _monitor.HOST_SERVER))
@app.route('/api/settings/route-health', methods=['POST'])
@csrf_protect
@login_required
def api_save_route_health():
data = request.get_json(silent=True) or {}
existing = load_settings()
enabled = bool(data['enabled']) if 'enabled' in data else existing.get('route_check_enabled', True)
try:
interval = int(data.get('interval', existing.get('route_check_interval', _rh.DEFAULT_INTERVAL)))
except (TypeError, ValueError):
return jsonify({'error': gettext('Interval must be a number of seconds')}), 400
if interval not in _rh.INTERVALS:
return jsonify({'error': gettext('Interval must be one of %(intervals)s seconds',
intervals=', '.join(str(i) for i in _rh.INTERVALS))}), 400
update_settings(route_check_enabled=enabled,
route_check_interval=interval)
return jsonify({'ok': True, 'enabled': enabled, 'interval': interval})
_CGNAT_NETWORK = ipaddress.ip_network('100.64.0.0/10')
def _classify_ip(ip: str) -> str:
try:
addr = ipaddress.ip_address((ip or '').strip())
except ValueError:
return 'unknown'
if addr.is_loopback:
return 'loopback'
if addr.is_link_local:
return 'link-local'
if addr.version == 4 and addr in _CGNAT_NETWORK:
return 'cgnat'
if addr.is_private:
return 'private'
return 'public'
@app.route('/api/diagnostics/client-ip')
@login_required
def api_client_ip_diagnostic():
orig = request.environ.get('werkzeug.proxy_fix.orig') or {}
socket_peer = orig.get('REMOTE_ADDR', '') or request.environ.get('REMOTE_ADDR', '') or ''
headers = {
'X-Forwarded-For': request.headers.get('X-Forwarded-For', ''),
'X-Real-IP': request.headers.get('X-Real-IP', ''),
'CF-Connecting-IP': request.headers.get('CF-Connecting-IP', ''),
'X-Forwarded-Proto': request.headers.get('X-Forwarded-Proto', ''),
'X-Forwarded-Host': request.headers.get('X-Forwarded-Host', ''),
}
xff_chain = [p.strip() for p in headers['X-Forwarded-For'].split(',') if p.strip()]
effective = request.remote_addr or ''
seen = [ip for ip in [effective, socket_peer, *xff_chain] if ip]
classes = {ip: _classify_ip(ip) for ip in seen}
return jsonify({
'effective_ip': effective,
'effective_class': _classify_ip(effective),
'socket_peer': socket_peer,
'socket_peer_class': _classify_ip(socket_peer),
'headers': headers,
'forwarded_for_chain': xff_chain,
'proxy_hops': PROXY_FIX_HOPS,
'proxy_trusted': env.peer_is_trusted(socket_peer),
'trusted_proxies': env.trusted_proxies_list(),
'classes': classes,
})
@app.route('/api/settings/backup-retention', methods=['POST'])
@csrf_protect
@login_required
def api_save_backup_retention():
try:
data = request.get_json() or {}
keep = max(0, int(data.get('backup_keep_count', 0)))
update_settings(backup_keep_count=keep)
return jsonify({'success': True, 'backup_keep_count': keep})
except (ValueError, TypeError):
return jsonify({'error': gettext('Invalid keep count')}), 400
except Exception as e:
logger.exception("Backup retention save error")
return jsonify({'error': str(e)}), 500
@app.route('/api/settings/self-route', methods=['GET'])
@login_required
def api_get_self_route():
settings = load_settings()
sr = settings.get('self_route', {'domain': '', 'service_url': ''})
default_ep = _best_entrypoint()
if not sr.get('domain'):
hostname = request.args.get('hostname', '').strip().lower()
if hostname:
found = _find_existing_self_route(hostname)
if found:
return jsonify({**found, 'default_entry_point': default_ep})
return jsonify({**sr, 'default_entry_point': default_ep})
@app.route('/api/settings/self-route', methods=['POST'])
@csrf_protect
@login_required
def api_save_self_route():
data = request.get_json(silent=True) or {}
domain = str(data.get('domain', '')).strip()
service_url = str(data.get('service_url', '')).strip() or 'http://traefik-manager:5000'
router_name = str(data.get('router_name', 'traefik-manager')).strip() or 'traefik-manager'
entry_point = str(data.get('entry_point', '')).strip() or _best_entrypoint()
settings = load_settings()
if domain:
_write_self_route(domain, service_url, settings.get('cert_resolver', 'cloudflare'), router_name, entry_point)
sr = {'domain': domain, 'service_url': service_url, 'router_name': router_name, 'entry_point': entry_point}
else:
_delete_self_route()
sr = {'domain': '', 'service_url': ''}
update_settings(self_route=sr)
return jsonify({'ok': True})
def _file_template_map(path):
if path and os.path.exists(path):
with open(path, 'r') as f:
_, mapping = _sanitize_go_templates(f.read())
return mapping
return {}
def _parse_backends_json(raw):
import json as _json
if not raw:
return None
try:
data = _json.loads(raw)
except (ValueError, TypeError):
return None
return data if isinstance(data, dict) else None
def _clean_priority(value):
try:
n = int(value)
except (TypeError, ValueError):
return None
return n if n != 0 else None
def _clean_duration(value):
v = str(value or '').strip()
if not v:
return ''
if v.isdigit():
return v + 's'
return v if re.match(r'^(\d+(ms|s|m|h))+$', v) else ''
_SCHEME_IN_HOST = re.compile(r'^(https?|h2c)://', re.I)
def _backend_servers(rows, key, scheme_default='http'):
servers = []
for row in rows if isinstance(rows, list) else []:
if not isinstance(row, dict):
continue
host = str(row.get('host') or '').strip()
if not host:
continue
port = str(row.get('port') or '').strip()
if key == 'url':
if _SCHEME_IN_HOST.match(host):
servers.append({'url': host})
else:
scheme = str(row.get('scheme') or scheme_default).strip() or scheme_default
servers.append({'url': f"{scheme}://{host}:{port}" if port else f"{scheme}://{host}"})
else:
servers.append({key: f"{host}:{port}" if port else host})
return servers
def _sticky_block(sticky):
if not isinstance(sticky, dict) or not sticky.get('enabled'):
return None
cookie = {}
name = str(sticky.get('cookieName') or '').strip()
if name:
cookie['name'] = name
if sticky.get('secure'):
cookie['secure'] = True
if sticky.get('httpOnly'):
cookie['httpOnly'] = True
return {'cookie': cookie}
_HC_DURATIONS = ('interval', 'unhealthyInterval', 'timeout')
_HC_TEXT = ('scheme', 'mode', 'hostname', 'method')
_HC_KNOWN = ('path', 'port', 'status', 'followRedirects', 'headers') + _HC_DURATIONS + _HC_TEXT
def _healthcheck_block(hc):
if not isinstance(hc, dict) or not hc.get('enabled'):
return None
path = str(hc.get('path') or '').strip()
block = {'path': path} if path else {}
for field in _HC_DURATIONS:
val = _clean_duration(hc.get(field))
if val:
block[field] = val
for field in _HC_TEXT:
val = str(hc.get(field) or '').strip()
if val:
block[field] = val
for field in ('port', 'status'):
val = str(hc.get(field) or '').strip()
if val.isdigit():
block[field] = int(val)
if hc.get('followRedirects') is False:
block['followRedirects'] = False
headers = hc.get('headers')
if isinstance(headers, dict):
clean = {str(k).strip(): str(v) for k, v in headers.items() if str(k).strip()}
if clean:
block['headers'] = clean
return block
def _streaming_forwarding_timeouts() -> dict:
return {'dialTimeout': '30s', 'responseHeaderTimeout': '0s', 'idleConnTimeout': '90s'}
def _headers_toggles_from_form(form) -> dict:
perms = {}
for feat in HEADERS_PRESET_FEATURES:
val = form.get(f'hp_perm_{feat}', '')
perms[feat] = val if val in ('self', 'all', 'block') else 'block'
return {
'perms': perms,
'hsts': form.get('hp_hsts') == 'true',
'nosniff': form.get('hp_nosniff') == 'true',
'frameDeny': form.get('hp_frameDeny') == 'true',
'referrer': (form.get('hp_referrer') or '').strip(),
}
def _composite_children(svc_def) -> list:
if not isinstance(svc_def, dict):
return []
names = []
for key in ('weighted', 'highestRandomWeight'):
block = svc_def.get(key)
if isinstance(block, dict):
for child in (block.get('services') or []):
if isinstance(child, dict):
names.append(child.get('name', ''))
mirroring = svc_def.get('mirroring')
if isinstance(mirroring, dict):
names.append(mirroring.get('service', ''))
for mirror in (mirroring.get('mirrors') or []):
if isinstance(mirror, dict):
names.append(mirror.get('name', ''))
failover = svc_def.get('failover')
if isinstance(failover, dict):
names.append(failover.get('service', ''))
names.append(failover.get('fallback', ''))
return [n for n in names if isinstance(n, str) and n]
def _service_in_disabled_snapshots(target: str, exclude_router: str) -> bool:
try:
disabled = load_settings().get('disabled_routes', {}) or {}
except Exception:
return False
for key, snap in disabled.items():
if not isinstance(snap, dict):
continue
if _disabled_router_name(str(key)) == exclude_router:
continue
router = snap.get('router')
if isinstance(router, dict) and _svc_key(router.get('service', '')) == target:
return True
for child in _composite_children(snap.get('service')):
if _svc_key(child) == target:
return True
return False
def _service_shared(config: dict, svc_name: str, exclude_router: str) -> bool:
target = _svc_key(svc_name)
for sec in ('http', 'tcp', 'udp'):
sec_cfg = config.get(sec) or {}
routers = sec_cfg.get('routers') or {}
for rn, rd in routers.items():
if rn == exclude_router or not isinstance(rd, dict):
continue
if _svc_key(rd.get('service', '')) == target:
return True
for sn, sd in (sec_cfg.get('services') or {}).items():
if _svc_key(sn) == target:
continue
for child in _composite_children(sd):
if _svc_key(child) == target:
return True
return _service_in_disabled_snapshots(target, exclude_router)
def _disabled_key(disabled, full_id, plain_id, prefix=''):
for key in (full_id, plain_id, prefix + plain_id):
if key and key in disabled:
return key
for key in disabled:
if prefix and not key.startswith(prefix):
continue
if _disabled_router_name(key) == plain_id:
return key
return None
def _save_disabled_routes(settings, disabled):
_save_edit_dicts(disabled_routes=disabled)
def _toggle_route(route_id: str, enable: bool):
settings = load_settings()
disabled = settings.get('disabled_routes', {})
rname = route_id.split('::', 1)[1] if '::' in route_id else route_id
if enable:
if route_id not in disabled:
return
saved = disabled.pop(route_id)
proto = saved.get('protocol', 'http')
router = saved.get('router', {})
svc_name = _svc_key(router.get('service', rname))
svc = saved.get('service', {})
cf = saved.get('configFile', '')
svc_cf = saved.get('serviceConfigFile', cf)
def _resolve_or_fallback(f):
p = _resolve_config_path(f)
if not p and f:
safe = f if f.endswith(('.yml', '.yaml')) else f + '.yml'
candidate = os.path.join(os.path.dirname(env.CONFIG_PATH) or '.', safe)
p = candidate if _is_safe_path(candidate) else env.CONFIG_PATH
return p or env.CONFIG_PATH
target_path = _resolve_or_fallback(cf)
config = load_config(target_path)
config.setdefault(proto, {}).setdefault('routers', {})[rname] = router
if svc_cf == cf or not svc_cf:
if svc:
config.setdefault(proto, {}).setdefault('services', {})[svc_name] = svc
create_backup(target_path)
save_config(_strip_empty_sections(config), target_path)
else:
create_backup(target_path)
save_config(_strip_empty_sections(config), target_path)
svc_path = _resolve_or_fallback(svc_cf)
svc_config = load_config(svc_path)
if svc:
svc_config.setdefault(proto, {}).setdefault('services', {})[svc_name] = svc
create_backup(svc_path)
save_config(_strip_empty_sections(svc_config), svc_path)
else:
proto = None
router = None
svc_name = None
svc = None
target_path = None
svc_path = None
svc_config = None
cf_prefix = route_id.split('::', 1)[0] if '::' in route_id else ''
_pref_path = _resolve_config_path(cf_prefix) if cf_prefix else None
search_paths = [_pref_path] if _pref_path else env.CONFIG_PATHS
for p in search_paths:
config = load_config(p)
for prot in ('http', 'tcp', 'udp'):
routers = config.get(prot, {}).get('routers', {})
if rname in routers:
proto = prot
router = dict(routers.pop(rname))
svc_name = _svc_key(router.get('service', rname))
target_path = p
svc_config = config
break
if proto:
break
if proto is None:
return
if _service_shared(svc_config, svc_name, rname):
svc = dict(svc_config.get(proto, {}).get('services', {}).get(svc_name, {}))
else:
svc = dict(svc_config.get(proto, {}).get('services', {}).pop(svc_name, {}))
if not svc:
for p in env.CONFIG_PATHS:
if p == target_path:
continue
other = load_config(p)
if svc_name in other.get(proto, {}).get('services', {}):
svc = dict(other[proto]['services'].pop(svc_name))
svc_path = p
svc = _restore_go_templates(svc, _file_template_map(p))
other_stripped = _strip_empty_sections(other)
create_backup(p)
save_config(other_stripped, p)
break
_target_map = _file_template_map(target_path)
router = _restore_go_templates(router, _target_map)
if svc_path is None:
svc = _restore_go_templates(svc, _target_map)
cf = os.path.basename(target_path) if (env.MULTI_CONFIG or ACTIVE_CONFIG_DIR) else ''
svc_cf = os.path.basename(svc_path) if svc_path and (env.MULTI_CONFIG or ACTIVE_CONFIG_DIR) else cf
disabled[route_id] = {'protocol': proto, 'router': router, 'service': svc, 'configFile': cf, 'serviceConfigFile': svc_cf}
create_backup(target_path)
save_config(_strip_empty_sections(svc_config), target_path)
_save_edit_dicts(disabled_routes=disabled)
def _self_route_service_name() -> str:
try:
return str((load_settings().get('self_route') or {}).get('router_name')
or 'traefik-manager')
except Exception:
return 'traefik-manager'
def _collect_file_services(configs):
out = {'http': set(), 'tcp': set(), 'udp': set()}
skip = _self_route_service_name()
for cfg in configs:
for sec in out:
out[sec].update(k for k in ((cfg.get(sec) or {}).get('services') or {})
if isinstance(k, str) and k and k != skip)
return {sec: sorted(names) for sec, names in out.items()}
@app.route('/api/routes')
@login_required
def api_routes():
apps, middlewares = _build_all_apps(include_external=False, live_tls=True)
return jsonify({'apps': apps, 'middlewares': middlewares,
'configErrors': _get_config_parse_errors(),
'services': _collect_file_services(load_config(_p) for _p in env.CONFIG_PATHS)})
@app.route('/api/routes/all')
@login_required
def api_routes_all():
apps, middlewares = _build_all_apps(include_external=True, include_internal=True)
services = _collect_file_services(load_config(_p) for _p in env.CONFIG_PATHS)
return jsonify({'apps': apps, 'middlewares': middlewares, 'services': services})
@app.route('/api/configs')
@login_required
def api_configs():
env.refresh_config_paths()
return jsonify({
'files': [{'label': os.path.basename(p), 'path': p} for p in env.CONFIG_PATHS],
'configDirSet': bool(ACTIVE_CONFIG_DIR),
})
def _read_groups_file():
if not os.path.exists(GROUPS_CONFIG_FILE):
return {}
try:
_y = SafeYAML(typ='safe')
with open(GROUPS_CONFIG_FILE, 'r') as f:
return _y.load(f) or {}
except Exception:
logger.exception("Failed to read dashboard config")
return {}
def _groups_scope_key(server):
s = str(server or '').strip()
return s if s and s != 'host' else ''
def _read_groups_config(server=''):
data = _read_groups_file()
key = _groups_scope_key(server)
if key:
data = ((data.get('servers') or {}).get(key)) or {}
return {
'custom_groups': list(data.get('custom_groups', []) or []),
'route_overrides': _sanitize_route_overrides(data.get('route_overrides', {}) or {}),
}
def _sanitize_route_overrides(overrides):
out = {}
for rid, ov in (overrides or {}).items():
if not isinstance(ov, dict):
continue
ov = dict(ov)
url = str(ov.get('url') or '').strip()
if not url or not url.lower().startswith(('http://', 'https://')):
ov.pop('url', None)
else:
ov['url'] = url
out[rid] = ov
return out
def _write_groups_config(data, server=''):
with _locks.file_lock(GROUPS_CONFIG_FILE):
doc = _read_groups_file()
scope = {
'custom_groups': list(data.get('custom_groups', []) or []),
'route_overrides': _sanitize_route_overrides(data.get('route_overrides', {})),
}
key = _groups_scope_key(server)
if key:
servers = dict(doc.get('servers') or {})
servers[key] = scope
doc['servers'] = servers
else:
doc['custom_groups'] = scope['custom_groups']
doc['route_overrides'] = scope['route_overrides']
_y = SafeYAML(typ='safe')
tmp = f"{GROUPS_CONFIG_FILE}.tmp.{os.getpid()}.{threading.get_ident()}"
try:
with open(tmp, 'w') as f:
_y.dump(doc, f)
_cfg._replace_or_copy(tmp, GROUPS_CONFIG_FILE)
finally:
if os.path.exists(tmp):
try:
os.unlink(tmp)
except OSError:
pass
@app.route('/api/dashboard/config', methods=['GET'])
@login_required
def dashboard_config_get():
cfg = _read_groups_config(request.args.get('server', ''))
sr = load_settings().get('self_route', {})
cfg['tm_route_name'] = sr.get('router_name', 'traefik-manager') or 'traefik-manager'
return jsonify(cfg)
@app.route('/api/dashboard/config', methods=['POST'])
@login_required
@csrf_protect
def dashboard_config_post():
data = request.get_json() or {}
_write_groups_config(data, data.get('server', '') or request.args.get('server', ''))
return jsonify({'ok': True})
@app.route('/api/dashboard/icon/<slug>')
@login_required
def dashboard_icon(slug):
slug = re.sub(r'[^a-z0-9-]', '', slug.lower())
if not slug:
return ('', 404)
cache_path = os.path.join(GROUPS_CACHE_DIR, slug + '.png')
miss_path = os.path.join(GROUPS_CACHE_DIR, slug + '.404')
if os.path.exists(cache_path):
return send_file(cache_path, mimetype='image/png', max_age=86400, conditional=True)
if os.path.exists(miss_path):
return ('', 404)
try:
r = requests.get(f'https://cdn.jsdelivr.net/gh/selfhst/icons@main/png/{slug}.png', timeout=2)
if r.status_code == 200 and 'image' in r.headers.get('content-type', ''):
with open(cache_path, 'wb') as wf:
wf.write(r.content)
return send_file(cache_path, mimetype='image/png', max_age=86400, conditional=True)
open(miss_path, 'w').close()
except Exception:
pass
return ('', 404)
def _toggle_route_agent(agent: dict, agent_id: str, route_id: str, enable: bool):
settings = load_settings()
disabled = dict(settings.get('disabled_routes', {}))
rname = route_id.split('::', 1)[1] if '::' in route_id else route_id
store_key = f"agent_{agent_id}::{route_id}"
if enable:
if store_key not in disabled:
return
saved = disabled.pop(store_key)
proto = saved.get('protocol', 'http')
router = saved.get('router', {})
svc_name = _svc_key(router.get('service', rname))
svc = saved.get('service', {})
cf = saved.get('configFile', 'dynamic.yml')
all_cfgs = _agent_load_configs(agent)
config = all_cfgs.get(cf, {})
config.setdefault(proto, {}).setdefault('routers', {})[rname] = router
if svc:
config.setdefault(proto, {}).setdefault('services', {})[svc_name] = svc
_agent_write_config(agent, cf, config)
else:
all_cfgs = _agent_load_configs(agent)
cf_prefix = route_id.split('::', 1)[0] if '::' in route_id else ''
cfg_items = [(cf_prefix, all_cfgs[cf_prefix])] if cf_prefix in all_cfgs else list(all_cfgs.items())
for fname, config in cfg_items:
for prot in ('http', 'tcp', 'udp'):
routers = config.get(prot, {}).get('routers', {})
if rname in routers:
router = dict(routers.pop(rname))
svc_name = _svc_key(router.get('service', rname))
if _service_shared(config, svc_name, rname):
svc = dict(config.get(prot, {}).get('services', {}).get(svc_name, {}))
else:
svc = dict(config.get(prot, {}).get('services', {}).pop(svc_name, {}))
disabled[store_key] = {'protocol': prot, 'router': router, 'service': svc, 'configFile': fname}
_agent_write_config(agent, fname, config)
break
else:
continue
break
_save_edit_dicts(disabled_routes=disabled)
@app.route('/api/routes/<path:route_id>/toggle', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_toggle_route(route_id):
body = request.get_json(force=True, silent=True) or {}
enable = body.get('enable', True)
agent_id = body.get('agent_id', '').strip()
agent = _agent_by_id(agent_id) if agent_id else None
try:
if agent:
_toggle_route_agent(agent, agent_id, route_id, bool(enable))
threading.Thread(target=lambda: _git_push_agent_if_enabled(agent, 'route toggle'), daemon=True).start()
else:
_toggle_route(route_id, bool(enable))
threading.Thread(target=lambda: _git_push_if_enabled('route toggle'), daemon=True).start()
return jsonify({'ok': True})
except Exception as e:
logger.error(f"Toggle route error: {e}")
return jsonify({'ok': False, 'message': str(e)}), 500
def _route_transport_name(svc):
lb = svc.get('loadBalancer') if isinstance(svc, dict) else None
return str(lb.get('serversTransport') or '') if isinstance(lb, dict) else ''
def _local_middleware_names(router):
names = []
for raw in _to_list(router.get('middlewares')):
name = str(raw or '')
provider = name.split('@', 1)[1] if '@' in name else ''
if provider and provider != 'file':
continue
short = name.split('@', 1)[0]
if short and short not in names:
names.append(short)
return names
def _definition_section(config, scope, kind):
if not isinstance(config, dict):
return {}
section = config.get(scope)
if not isinstance(section, dict):
return {}
holder = section.get(kind)
return holder if isinstance(holder, dict) else {}
def _route_dependency_names(proto, router, svc):
wanted = []
transport = _route_transport_name(svc)
if transport:
wanted.append((proto, 'serversTransports', transport))
for name in _local_middleware_names(router):
wanted.append((proto, 'middlewares', name))
tls = router.get('tls')
option = str((tls or {}).get('options') or '') if isinstance(tls, dict) else ''
option = option.split('@', 1)[0]
if option:
wanted.append(('tls', 'options', option))
return wanted
def _find_definition(scope, kind, name, own_config, own_path):
defined = _definition_section(own_config, scope, kind).get(name)
if defined is not None:
return defined, own_path
for path in env.CONFIG_PATHS:
if own_path and os.path.abspath(path) == os.path.abspath(own_path):
continue
try:
other = load_config(path)
except Exception:
logger.debug(f"Cannot read {path} while locating {kind}/{name}", exc_info=True)
continue
defined = _definition_section(other, scope, kind).get(name)
if defined is not None:
return defined, path
return None, None
def _add_route_dependencies(out, config, proto, router, svc, own_path=None):
origins = {}
for scope, kind, name in _route_dependency_names(proto, router, svc):
defined, path = _find_definition(scope, kind, name, config, own_path)
if defined is None:
continue
out.setdefault(scope, {}).setdefault(kind, {})[name] = dict(defined)
origins.setdefault(scope, {}).setdefault(kind, {})[name] = os.path.basename(path)
return origins
def _plain_value(value, templates=None):
if isinstance(value, dict):
return {str(k): _plain_value(v, templates) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [_plain_value(v, templates) for v in value]
if isinstance(value, str) and templates:
for placeholder, original in templates.items():
value = value.replace(placeholder, original)
return value
_ELSEWHERE_MESSAGES = {
'readonly': lambda name, where: gettext(
'%(name)s is defined in %(file)s, which is read-only here.', name=name, file=where),
'stale': lambda where: gettext(
'%(file)s changed on disk since this editor opened. Reopen it and make the change again.',
file=where),
'renamed': lambda name, where: gettext(
'Renaming %(name)s here would leave it behind in %(file)s. Rename it on the Middlewares tab.',
name=name, file=where),
}
def _dependency_scopes(data):
for proto in ('http', 'tcp', 'udp'):
for kind in ('serversTransports', 'middlewares'):
if _definition_section(data, proto, kind):
yield proto, kind
if _definition_section(data, 'tls', 'options'):
yield 'tls', 'options'
def _sections_defined_elsewhere(new_data, config, target_path, user_map):
unchanged, changed = [], []
for scope, kind in _dependency_scopes(new_data):
for name, edited in _definition_section(new_data, scope, kind).items():
defined, path = _find_definition(scope, kind, name, config, target_path)
if path is None or os.path.abspath(path) == os.path.abspath(target_path):
continue
with open(path, 'r') as f:
_, their_map = _sanitize_go_templates(f.read())
here = _plain_value(edited, user_map)
there = _plain_value(defined, their_map)
(unchanged if here == there else changed).append(
{'scope': scope, 'kind': kind, 'name': name, 'path': path,
'file': os.path.basename(path), 'value': edited})
return unchanged, changed
_MISSING_REFERENCE_MESSAGES = {
'middlewares': lambda name: gettext(
'The middleware %(name)s is not defined anywhere. Create it first, or correct the name.',
name=name),
'services': lambda name: gettext(
'The service %(name)s is not defined anywhere. Create it first, or correct the name.',
name=name),
'serversTransports': lambda name: gettext(
'The serversTransport %(name)s is not defined anywhere. Create it first, or correct the name.',
name=name),
'options': lambda name: gettext(
'The TLS options %(name)s are not defined anywhere. Create them first, or correct the name.',
name=name),
}
def _file_reference(raw):
name = str(raw or '').strip()
if '@' not in name:
return name
short, provider = name.split('@', 1)
return short if provider == 'file' else ''
def _reference_exists(new_data, config, target_path, scope, kind, name):
if name in _definition_section(new_data, scope, kind):
return True
_defined, path = _find_definition(scope, kind, name, config, target_path)
return path is not None
def _missing_route_references(new_data, config, target_path, untouched=()):
missing = []
skip = {(c['scope'], c['kind'], c['name']) for c in untouched}
def check(scope, kind, raw):
name = _file_reference(raw)
if not name or (kind == 'options' and name == 'default'):
return
if not _reference_exists(new_data, config, target_path, scope, kind, name):
if (kind, name) not in missing:
missing.append((kind, name))
for proto in ('http', 'tcp', 'udp'):
for rname, router in _definition_section(new_data, proto, 'routers').items():
if not isinstance(router, dict):
continue
for raw in _to_list(router.get('middlewares')):
check(proto, 'middlewares', raw)
check(proto, 'services', router.get('service', rname))
tls = router.get('tls')
if isinstance(tls, dict):
check('tls', 'options', tls.get('options'))
for svc in _definition_section(new_data, proto, 'services').values():
check(proto, 'serversTransports', _route_transport_name(svc))
for mw_name, mw in _definition_section(new_data, proto, 'middlewares').items():
if not isinstance(mw, dict) or (proto, 'middlewares', mw_name) in skip:
continue
chain = mw.get('chain')
if isinstance(chain, dict):
for raw in _to_list(chain.get('middlewares')):
check(proto, 'middlewares', raw)
errors = mw.get('errors')
if isinstance(errors, dict):
check(proto, 'services', errors.get('service'))
return missing
def _definition_reference_warnings(out, origins, config, target_path):
warnings = []
for proto in ('http', 'tcp', 'udp'):
where = (origins.get(proto) or {}).get('middlewares') or {}
for name, mw in _definition_section(out, proto, 'middlewares').items():
if not isinstance(mw, dict):
continue
refs = []
chain = mw.get('chain')
if isinstance(chain, dict):
refs += [(proto, 'middlewares', r) for r in _to_list(chain.get('middlewares'))]
errors = mw.get('errors')
if isinstance(errors, dict):
refs.append((proto, 'services', errors.get('service')))
for scope, kind, raw in refs:
missing = _file_reference(raw)
if not missing or _reference_exists(out, config, target_path, scope, kind, missing):
continue
warnings.append({'name': name, 'kind': kind, 'missing': missing,
'file': where.get(name, os.path.basename(target_path))})
return warnings
def _file_fingerprint(path):
try:
with open(path, 'rb') as f:
return hashlib.sha256(f.read()).hexdigest()[:16]
except OSError:
return ''
def _route_dependency_fingerprints(origins, own_path):
prints = {os.path.basename(own_path): _file_fingerprint(own_path)}
for kinds in origins.values():
for names in kinds.values():
for where in names.values():
if where in prints:
continue
for path in env.CONFIG_PATHS:
if os.path.basename(path) == where:
prints[where] = _file_fingerprint(path)
break
return prints
def _routes_using(scope, kind, name):
users = []
services = set()
if kind == 'serversTransports':
for path in env.CONFIG_PATHS:
try:
cfg = load_config(path)
except Exception:
continue
for svc_name, svc in _definition_section(cfg, scope, 'services').items():
if _route_transport_name(svc) == name:
services.add(str(svc_name))
for path in env.CONFIG_PATHS:
try:
cfg = load_config(path)
except Exception:
continue
for proto in ('http', 'tcp', 'udp'):
if kind != 'options' and proto != scope:
continue
for rname, router in _definition_section(cfg, proto, 'routers').items():
if not isinstance(router, dict):
continue
if kind == 'middlewares':
hit = name in _local_middleware_names(router)
elif kind == 'serversTransports':
hit = _svc_key(router.get('service', rname)) in services
else:
tls = router.get('tls')
option = str((tls or {}).get('options') or '') if isinstance(tls, dict) else ''
hit = option.split('@', 1)[0] == name
if hit:
users.append(str(rname))
return sorted(set(users))
def _shared_change_prompt(changes):
out = []
for change in changes:
routes = _routes_using(change['scope'], change['kind'], change['name'])
out.append({'name': change['name'], 'kind': change['kind'], 'file': change['file'],
'usedBy': {'count': len(routes), 'routes': routes[:3]}})
return out
def _atomic_write(path, text):
tmp = f"{path}.tmp.{os.getpid()}.{threading.get_ident()}"
try:
with open(tmp, 'w') as f:
f.write(text)
f.flush()
os.fsync(f.fileno())
_cfg._replace_or_copy(tmp, path)
finally:
try:
os.unlink(tmp)
except OSError:
pass
def _renamed_copy(new_data, config, target_path, route_name, user_map):
for proto in ('http', 'tcp'):
old_routers = _definition_section(config, proto, 'routers')
for name, new_router in _definition_section(new_data, proto, 'routers').items():
old_router = old_routers.get(name)
if not isinstance(old_router, dict) or not isinstance(new_router, dict):
continue
old_svc = _definition_section(config, proto, 'services').get(
_svc_key(old_router.get('service', name)))
new_svc = _definition_section(new_data, proto, 'services').get(
_svc_key(new_router.get('service', name)))
kept = set(_route_dependency_names(proto, new_router, new_svc))
for scope, kind, dropped in _route_dependency_names(proto, old_router, old_svc):
if (scope, kind, dropped) in kept or kind not in ('middlewares', 'serversTransports'):
continue
old_body, old_path = _find_definition(scope, kind, dropped, config, target_path)
if old_body is None:
continue
with open(old_path, 'r') as f:
_, old_map = _sanitize_go_templates(f.read())
for candidate, body in _definition_section(new_data, scope, kind).items():
if candidate == dropped:
continue
_found, where = _find_definition(scope, kind, candidate, config, target_path)
if where is not None:
continue
if _plain_value(body, user_map) != _plain_value(old_body, old_map):
continue
others = [r for r in _routes_using(scope, kind, dropped) if r != route_name]
return {'from': dropped, 'to': candidate, 'kind': kind,
'file': os.path.basename(old_path),
'usedBy': {'count': len(others), 'routes': others[:3]}}
return None
def _write_shared_definitions(changes, user_map):
for path in sorted({c['path'] for c in changes}):
cfg = load_config(path)
with open(path, 'r') as f:
_, their_map = _sanitize_go_templates(f.read())
for change in changes:
if change['path'] != path:
continue
holder = cfg.setdefault(change['scope'], {})
holder.setdefault(change['kind'], {})[change['name']] = change['value']
create_backup(path)
stream = StringIO()
yaml.dump(_strip_empty_sections(cfg), stream)
text = stream.getvalue()
for placeholder, original in {**their_map, **user_map}.items():
text = text.replace(placeholder, original)
_atomic_write(path, text)
logger.info(f"Shared definitions written to {path} from a route raw save")
def _renamed_shared_definition(new_data, config, target_path):
for scope, kind in (('http', 'middlewares'), ('tcp', 'middlewares'),
('http', 'serversTransports'), ('tcp', 'serversTransports')):
present = _definition_section(new_data, scope, kind)
routers = _definition_section(new_data, scope, 'routers')
for rname, router in routers.items():
if not isinstance(router, dict):
continue
svc = _definition_section(new_data, scope, 'services').get(
_svc_key(router.get('service', rname)))
for dep_scope, dep_kind, dep_name in _route_dependency_names(scope, router, svc):
if dep_kind != kind or dep_scope != scope or dep_name in present:
continue
_defined, path = _find_definition(dep_scope, dep_kind, dep_name, config,
target_path)
if path is None or os.path.abspath(path) == os.path.abspath(target_path):
continue
for candidate in present:
_found, where = _find_definition(scope, kind, candidate, config, target_path)
if where is None:
return dep_name, os.path.basename(path)
return None, None
@app.route('/api/routes/<path:route_id>/raw', methods=['GET'])
@login_required
def api_route_raw_get(route_id):
rname = route_id.split('::', 1)[1] if '::' in route_id else route_id
cf = route_id.split('::', 1)[0] if '::' in route_id else ''
target_path = _resolve_config_path(cf) if cf else None
paths_to_scan = [target_path] if target_path else env.CONFIG_PATHS
for p in paths_to_scan:
config = load_config(p)
for proto in ('http', 'tcp', 'udp'):
routers = config.get(proto, {}).get('routers', {})
if rname in routers:
router = routers[rname]
svc_name = router.get('service', rname)
svc_key = _svc_key(svc_name)
svc = config.get(proto, {}).get('services', {}).get(svc_key)
out = {proto: {'routers': {rname: dict(router)}}}
if svc is not None:
out[proto]['services'] = {svc_name: dict(svc)}
origins = _add_route_dependencies(out, config, proto, router, svc, p)
stream = StringIO()
yaml.dump(out, stream)
raw = stream.getvalue()
with open(p, 'r') as f:
_, template_map = _sanitize_go_templates(f.read())
for placeholder, original in template_map.items():
raw = raw.replace(placeholder, original)
return jsonify({'raw': raw, 'configFile': os.path.basename(p), 'proto': proto,
'origins': origins,
'fingerprints': _route_dependency_fingerprints(origins, p),
'warnings': _definition_reference_warnings(out, origins, config, p)})
return jsonify({'error': gettext('Route not found')}), 404
@app.route('/api/routes/<path:route_id>/raw', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def api_route_raw_save(route_id):
body = request.get_json(force=True, silent=True) or {}
content = body.get('content', '')
if not content.strip():
return jsonify({'ok': False, 'error': gettext('No content')}), 400
rname = route_id.split('::', 1)[1] if '::' in route_id else route_id
cf = route_id.split('::', 1)[0] if '::' in route_id else ''
user_map = {}
user_counter = [0]
def _replace_user(m):
key = f'__TM_USER_{user_counter[0]}__'
user_map[key] = m.group(0)
user_counter[0] += 1
return key
sanitized_user = re.sub(r'\{\{[^}]*\}\}', _replace_user, content)
try:
new_data = yaml.load(sanitized_user)
if not isinstance(new_data, dict):
raise ValueError("Expected a YAML mapping")
except Exception as e:
return jsonify({'ok': False, 'error': gettext('Invalid YAML: %(error)s', error=e)}), 400
target_path = _resolve_config_path(cf) if cf else None
if not target_path:
for p in env.CONFIG_PATHS:
cfg = load_config(p)
for proto in ('http', 'tcp', 'udp'):
if rname in cfg.get(proto, {}).get('routers', {}):
target_path = p
break
if target_path:
break
if not target_path:
return jsonify({'ok': False, 'error': gettext('Route not found')}), 404
config = load_config(target_path)
file_map = {}
if os.path.exists(target_path):
with open(target_path, 'r') as f:
_, file_map = _sanitize_go_templates(f.read())
renamed, renamed_file = _renamed_shared_definition(new_data, config, target_path)
if renamed:
return jsonify({'ok': False, 'renamed': renamed,
'error': _ELSEWHERE_MESSAGES['renamed'](renamed, renamed_file)}), 409
copied = _renamed_copy(new_data, config, target_path, rname, user_map)
if copied and not body.get('applyRename'):
return jsonify({'ok': False, 'renamedCopy': copied}), 409
elsewhere, changed = _sections_defined_elsewhere(new_data, config, target_path, user_map)
for change in changed:
if not os.access(change['path'], os.W_OK):
return jsonify({'ok': False, 'readOnly': change['file'],
'error': _ELSEWHERE_MESSAGES['readonly'](change['name'],
change['file'])}), 409
sent = body.get('fingerprints') or {}
for change in changed:
was = sent.get(change['file'])
if was and was != _file_fingerprint(change['path']):
return jsonify({'ok': False, 'stale': change['file'],
'error': _ELSEWHERE_MESSAGES['stale'](change['file'])}), 409
missing = _missing_route_references(new_data, config, target_path, elsewhere)
if missing:
kind, name = missing[0]
return jsonify({'ok': False, 'error': _MISSING_REFERENCE_MESSAGES[kind](name),
'missingReferences': [{'kind': k, 'name': n} for k, n in missing]}), 409
if changed and not body.get('applyShared'):
return jsonify({'ok': False, 'needsConfirm': True,
'sharedChanges': _shared_change_prompt(changed)}), 409
for change in elsewhere:
_definition_section(new_data, change['scope'], change['kind']).pop(change['name'], None)
for change in changed:
_definition_section(new_data, change['scope'], change['kind']).pop(change['name'], None)
for proto in ('http', 'tcp', 'udp'):
proto_cfg = config.get(proto, {})
old_router = proto_cfg.get('routers', {}).pop(rname, None)
if old_router:
old_svc = _svc_key(old_router.get('service', rname))
proto_cfg.get('services', {}).pop(old_svc, None)
for proto in ('http', 'tcp', 'udp'):
new_proto = new_data.get(proto, {})
if not new_proto:
continue
section = config.setdefault(proto, {})
new_routers = new_proto.get('routers', {})
new_services = new_proto.get('services', {})
if new_routers:
section.setdefault('routers', {}).update(new_routers)
if new_services:
section.setdefault('services', {}).update(new_services)
for name in ('serversTransports', 'middlewares'):
edited = new_proto.get(name) or {}
if edited:
section.setdefault(name, {}).update(edited)
new_tls_options = (new_data.get('tls') or {}).get('options') or {}
if new_tls_options:
config.setdefault('tls', {}).setdefault('options', {}).update(new_tls_options)
try:
if changed:
_write_shared_definitions(changed, user_map)
create_backup(target_path)
combined_map = {**file_map, **user_map}
stream = StringIO()
yaml.dump(_strip_empty_sections(config), stream)
yaml_content = stream.getvalue()
for placeholder, original in combined_map.items():
yaml_content = yaml_content.replace(placeholder, original)
tmp = f"{target_path}.tmp.{os.getpid()}.{threading.get_ident()}"
try:
with open(tmp, 'w') as f:
f.write(yaml_content)
f.flush()
os.fsync(f.fileno())
_cfg._replace_or_copy(tmp, target_path)
finally:
try:
os.unlink(tmp)
except OSError:
pass
logger.info(f"Route '{rname}' raw config saved: {target_path}")
add_notification('success', _i18n.lazy_gettext("Route '%(name)s' updated", name=rname))
threading.Thread(target=lambda: _git_push_if_enabled('route raw save'), daemon=True).start()
return jsonify({'ok': True})
except Exception as e:
logger.exception("Route raw save error")
return jsonify({'ok': False, 'error': str(e)}), 500
def _static_cert_resolvers():
path = _get_static_config_path()
if not path or not os.path.exists(path):
return []
try:
with open(path, 'r') as f:
data = _yaml_safe.load(f) or {}
resolvers = data.get('certificatesResolvers') or {}
if isinstance(resolvers, dict):
return [str(k).strip() for k in resolvers if str(k).strip()]
except Exception:
logger.debug("Failed to read certificatesResolvers from static config", exc_info=True)
return []
@app.route('/')
@login_required
def index():
settings = load_settings()
apps, middlewares = _build_all_apps(include_external=False)
auth_on = _auth_required()
_ack = bool(load_settings().get('auth_external_ack'))
no_auth = not _auth_required() and not _ack
login_time = session.get('login_time', '')
config_paths_list = [{'label': os.path.basename(p), 'path': p} for p in env.CONFIG_PATHS]
cert_resolvers = [r.strip() for r in settings['cert_resolver'].split(',') if r.strip()]
for r in _static_cert_resolvers():
if r not in cert_resolvers:
cert_resolvers.append(r)
return render_template('index.html', apps=apps, domains=settings['domains'],
middlewares=middlewares, settings=settings,
auth_enabled=auth_on, no_auth=no_auth, login_time=login_time,
multi_config=env.MULTI_CONFIG,
config_paths_list=config_paths_list,
config_dir_set=bool(ACTIVE_CONFIG_DIR),
cert_resolvers=cert_resolvers,
crowdsec_enabled=bool(_cs_lapi_url() and (_cs_api_key() or _cs_has_machine())),
ui_prefs=settings.get('ui_prefs', {}))
def _is_fetch():
return request.headers.get('X-Requested-With') == 'fetch'
_SNI_NAME_RE = re.compile(r'^(?=.{1,253}$)[A-Za-z0-9_*]([A-Za-z0-9_.-]*[A-Za-z0-9])?$')
@app.route('/save', methods=['POST'])
@csrf_protect
@login_required
@_config_edit
def save_entry():
fetch = _is_fetch()
try:
settings = load_settings()
svc_name = request.form.get('serviceName', '').strip()
subdomain = request.form.get('subdomain', '').strip()
domain = request.form.get('domain', settings['domains'][0]).strip()
middlewares_in = request.form.get('middlewares', '').strip()
protocol = request.form.get('protocol', 'http').strip().lower()
is_edit = request.form.get('isEdit') == 'true'
original_id = request.form.get('originalId', '')
tcp_rule = request.form.get('tcpRule', '').strip()
http_rule = request.form.get('httpRule', '').strip()
scheme = request.form.get('scheme', 'http').strip().lower()
pass_host = request.form.get('passHostHeader') == 'true'
_all_eps = request.form.getlist('entryPoints')
http_eps = [ep.strip() for ep in (_all_eps[0] if _all_eps else 'https').split(',') if ep.strip()] or ['https']
tcp_eps = [ep.strip() for ep in (_all_eps[1] if len(_all_eps) > 1 else '').split(',') if ep.strip()]
_all_ips = request.form.getlist('targetIp')
_all_ports = request.form.getlist('targetPort')
if protocol == 'tcp':
target_ip = (_all_ips[1] if len(_all_ips) > 1 else '').strip()
target_port = (_all_ports[1] if len(_all_ports) > 1 else '').strip()
elif protocol == 'udp':
target_ip = (_all_ips[2] if len(_all_ips) > 2 else '').strip()
target_port = (_all_ports[2] if len(_all_ports) > 2 else '').strip()
else:
target_ip = (_all_ips[0] if _all_ips else '').strip()
target_port = (_all_ports[0] if _all_ports else '').strip()
resolvers = [r.strip() for r in settings['cert_resolver'].split(',') if r.strip()]
_all_resolvers = request.form.getlist('certResolver')
cert_resolver_raw = (_all_resolvers[0] if _all_resolvers else '').strip()
no_tls = cert_resolver_raw == '__disabled__'
cert_resolver = '' if (cert_resolver_raw in ('__none__', 'none', '__disabled__')) else (cert_resolver_raw or (resolvers[0] if resolvers else ''))
use_tls_tcp = request.form.get('useTls') == 'true'
tls_passthrough = request.form.get('tlsPassthrough') == 'true'
mws_tcp_raw = request.form.get('middlewaresTcp')
tcp_cert_raw = (_all_resolvers[1] if len(_all_resolvers) > 1 else '').strip()
tcp_cert_resolver = '' if (tcp_cert_raw in ('__none__', 'none')) else (tcp_cert_raw or (resolvers[0] if resolvers else ''))
config_file_raw = request.form.get('configFile', '').strip()
agent_id = request.form.get('agent_id', '').strip()
agent = _agent_by_id(agent_id) if agent_id else None
target_path = None if agent else (_resolve_config_path(config_file_raw) or env.CONFIG_PATH)
cfg_filename = config_file_raw or 'dynamic.yml'
if not svc_name:
if fetch:
return jsonify({'ok': False, 'message': gettext('Service name is required')}), 400
flash(gettext('Service name is required'), "error")
return redirect(url_for('index'))
svc_name_err = _naming.name_error(svc_name)
if svc_name_err:
if fetch:
return jsonify({'ok': False, 'message': svc_name_err}), 400
flash(svc_name_err, "error")
return redirect(url_for('index'))
if protocol not in ('http', 'tcp', 'udp'):
if fetch:
return jsonify({'ok': False, 'message': gettext('Invalid protocol')}), 400
flash(gettext('Invalid protocol'), "error")
return redirect(url_for('index'))
_backends_field = {'http': 'backendsJsonHttp', 'tcp': 'backendsJsonTcp',
'udp': 'backendsJsonUdp'}[protocol]
_has_backends_json = bool((_parse_backends_json(request.form.get(_backends_field)) or {}).get('servers'))
service_ref = request.form.get('serviceRef', '').strip()
_service_ref_posted = 'serviceRef' in request.form
if _service_ref_posted and not service_ref:
_msg = gettext('Select a service to reference, or switch the backend to Manual.')
if fetch:
return jsonify({'ok': False, 'message': _msg}), 400
flash(_msg, "error")
return redirect(url_for('index'))
_existing_is_composite = _router_resolves_to_composite(
request.form.get('originalName', '').strip() or svc_name, agent)
if not target_ip and not _has_backends_json and not service_ref \
and not _existing_is_composite:
_msg = gettext('A backend host is required for %(protocol)s routes. '
'Send targetIp (repeated per protocol, index %(index)s) or %(field)s.',
protocol=protocol.upper(), index={'http': 0, 'tcp': 1, 'udp': 2}[protocol],
field=_backends_field)
if fetch:
return jsonify({'ok': False, 'message': _msg}), 400
flash(_msg, "error")
return redirect(url_for('index'))
if protocol in ('tcp', 'udp') and not _has_backends_json and not target_port and not service_ref:
_host, _sep, _tail = target_ip.rpartition(':')
if _sep and _tail.isdigit() and (':' not in _host or _host.endswith(']')):
target_ip, target_port = _host, _tail
else:
_msg = gettext('A backend port is required for %(protocol)s routes. '
'Send targetPort (repeated per protocol, index %(index)s) or %(field)s.',
protocol=protocol.upper(), index={'tcp': 1, 'udp': 2}[protocol],
field=_backends_field)
if fetch:
return jsonify({'ok': False, 'message': _msg}), 400
flash(_msg, "error")
return redirect(url_for('index'))
router_name = svc_name
service_name = f"{svc_name}-service"
if agent:
config = _agent_load_configs(agent).get(cfg_filename, {})
else:
create_backup(target_path)
config = load_config(target_path)
orig_parts = original_id.split('::', 1)
plain_original_id = orig_parts[1] if len(orig_parts) > 1 else original_id
orig_cfg_file = orig_parts[0] if len(orig_parts) > 1 else cfg_filename
_ledger = settings.get('managed_middlewares', {})
_ledger_changed = False
hdr_name = f"{router_name}-headers"
hdr_ledger_key = f"agent_{agent_id}::{hdr_name}" if agent else hdr_name
hdr_preset_present = protocol == 'http' and request.form.get('headersPresetPresent') == 'true'
hdr_preset_on = hdr_preset_present and request.form.get('headersPresetEnabled') == 'true'
hdr_preset_custom = request.form.get('headersPresetCustom') == 'true'
stream_preset_present = protocol == 'http' and request.form.get('streamingPresetPresent') == 'true'
stream_preset_on = stream_preset_present and request.form.get('streamingPresetEnabled') == 'true'
if hdr_preset_on:
_existing_hdr = config.get('http', {}).get('middlewares', {}).get(hdr_name)
_hdr_foreign = _existing_hdr is not None and (
hdr_ledger_key not in _ledger or (is_edit and orig_cfg_file != cfg_filename))
if _hdr_foreign:
_msg = gettext("A middleware named '%(name)s' already exists and wasn't created by the route presets. "
"Rename or remove it first, then re-save.", name=hdr_name)
if fetch:
return jsonify({'ok': False, 'message': _msg}), 409
flash(_msg, "error")
return redirect(url_for('index'))
_prev_tcp_mws = None
if is_edit and plain_original_id:
_prev_src = config
if orig_cfg_file != cfg_filename:
if agent:
_prev_src = _agent_load_configs(agent).get(orig_cfg_file, {})
else:
_prev_path = _resolve_config_path(orig_cfg_file)
_prev_src = load_config(_prev_path) if _prev_path else {}
_prev_tcp_mws = _prev_src.get('tcp', {}).get('routers', {}).get(plain_original_id, {}).get('middlewares')
for sec in ('http', 'tcp', 'udp'):
prev_router = _prev_src.get(sec, {}).get('routers', {}).get(plain_original_id)
if not isinstance(prev_router, dict):
continue
prev_svc = (prev_router.get('service') or '').split('@')[0].strip()
if prev_svc and prev_svc in _prev_src.get(sec, {}).get('services', {}):
service_name = prev_svc
break
if not service_ref and is_edit and plain_original_id:
for sec in ('http', 'tcp', 'udp'):
prev_router = _prev_src.get(sec, {}).get('routers', {}).get(plain_original_id)
if not isinstance(prev_router, dict):
continue
_prev_ref = (prev_router.get('service') or '').strip()
if _prev_ref and _prev_ref != f"{plain_original_id}-service":
_ref_in_svcs = _svc_key(_prev_ref) in (_prev_src.get(sec, {}).get('services') or {})
if not _ref_in_svcs or _service_shared(_prev_src, _prev_ref, plain_original_id):
service_ref = _prev_ref
break
if service_ref:
if _service_ref_posted and '@' not in service_ref:
if agent:
_ref_cfgs = list(_agent_load_configs(agent).values())
else:
_ref_cfgs = [load_config(_p) for _p in env.CONFIG_PATHS]
if not any(service_ref in ((_c.get(protocol) or {}).get('services') or {}) for _c in _ref_cfgs):
_msg = gettext("Service '%(service)s' does not exist for %(protocol)s routes.",
service=service_ref, protocol=protocol.upper())
if fetch:
return jsonify({'ok': False, 'message': _msg}), 400
flash(_msg, "error")
return redirect(url_for('index'))
service_name = service_ref
if is_edit and plain_original_id and orig_cfg_file != cfg_filename:
if agent:
old_all_cfgs = _agent_load_configs(agent)
old_config = old_all_cfgs.get(orig_cfg_file, {})
else:
orig_target_path = _resolve_config_path(orig_cfg_file)
old_config = load_config(orig_target_path) if orig_target_path else {}
for sec in ('http', 'tcp', 'udp'):
s = old_config.get(sec, {})
old_routers = s.get('routers', {})
old_svc = (old_routers.get(plain_original_id, {}).get('service') or '').strip()
if plain_original_id in old_routers:
del old_routers[plain_original_id]
if (old_svc and 'services' in s and old_svc in s['services']
and not service_ref
and not _service_shared(old_config, old_svc, plain_original_id)):
del s['services'][old_svc]
old_transport_name = f"{plain_original_id}-transport"
http_sec = old_config.get('http', {})
old_transports = http_sec.get('serversTransports', {})
old_tp_key = f"agent_{agent_id}::tp::{old_transport_name}" if agent else f"tp::{old_transport_name}"
if old_transport_name in old_transports and old_tp_key in _ledger:
del old_transports[old_transport_name]
if not old_transports:
del http_sec['serversTransports']
del _ledger[old_tp_key]
_ledger_changed = True
old_hdr_name = f"{plain_original_id}-headers"
old_hdr_key = f"agent_{agent_id}::{old_hdr_name}" if agent else old_hdr_name
if hdr_preset_present and old_hdr_key in _ledger:
old_hdr_mws = http_sec.get('middlewares', {})
old_hdr_mws.pop(old_hdr_name, None)
if not old_hdr_mws and 'middlewares' in http_sec:
del http_sec['middlewares']
del _ledger[old_hdr_key]
_ledger_changed = True
if agent:
_agent_write_config(agent, orig_cfg_file, old_config)
elif orig_target_path:
save_config(_strip_empty_sections(old_config), orig_target_path)
if is_edit and plain_original_id:
for sec in ('http', 'tcp', 'udp'):
s = config.get(sec, {})
old_routers = s.get('routers', {})
old_svc = (old_routers.get(plain_original_id, {}).get('service') or '').strip()
if plain_original_id != router_name and plain_original_id in old_routers:
del old_routers[plain_original_id]
if (old_svc and old_svc != service_name and 'services' in s and old_svc in s['services']
and not _service_shared(config, old_svc, plain_original_id)):
del s['services'][old_svc]
if hdr_preset_present and plain_original_id != router_name:
rn_hdr_name = f"{plain_original_id}-headers"
rn_hdr_key = f"agent_{agent_id}::{rn_hdr_name}" if agent else rn_hdr_name
if rn_hdr_key in _ledger:
rn_hdr_mws = config.get('http', {}).get('middlewares', {})
rn_hdr_mws.pop(rn_hdr_name, None)
if not rn_hdr_mws and 'http' in config and 'middlewares' in config['http']:
del config['http']['middlewares']
del _ledger[rn_hdr_key]
_ledger_changed = True
if protocol == 'http':
if http_rule:
rule = http_rule
else:
selected_domains = request.form.getlist('domains') or [domain]
if subdomain and '.' in subdomain:
rule = f"Host(`{subdomain}`)"
elif subdomain:
hosts = [f"Host(`{subdomain}.{d}`)" for d in selected_domains]
rule = " || ".join(hosts)
else:
hosts = [f"Host(`{d}`)" for d in selected_domains]
rule = " || ".join(hosts)
if target_ip.startswith(('http://', 'https://')):
target_url = target_ip
else:
target_url = f"{scheme}://{target_ip}:{target_port}" if target_port else f"{scheme}://{target_ip}"
mws = [m.strip() for m in middlewares_in.split(',')] if middlewares_in else []
insecure = request.form.get('insecureSkipVerify') == 'true'
sni_posted = 'backendServerName' in request.form
sni_name = request.form.get('backendServerName', '').strip()
if sni_name and not _SNI_NAME_RE.match(sni_name):
_sni_msg = gettext('Backend server name must be a hostname, got %(value)s', value=repr(sni_name))
if fetch:
return jsonify({'ok': False, 'message': _sni_msg}), 400
flash(_sni_msg, "error")
return redirect(url_for('index'))
config.setdefault('http', {}).setdefault('routers', {})
config['http'].setdefault('services', {})
if hdr_preset_present:
if hdr_preset_on or hdr_ledger_key in _ledger:
mws = [m for m in mws if m != hdr_name]
if hdr_preset_on and not hdr_preset_custom:
config['http'].setdefault('middlewares', {})[hdr_name] = _build_headers_middleware(_headers_toggles_from_form(request.form))
_hdr_present_now = config.get('http', {}).get('middlewares', {}).get(hdr_name) is not None
if hdr_preset_on and _hdr_present_now:
mws.append(hdr_name)
if hdr_ledger_key not in _ledger:
_ledger[hdr_ledger_key] = {'kind': 'route-headers', 'route': router_name}
_ledger_changed = True
elif hdr_ledger_key in _ledger:
_hdr_sec = config.get('http', {}).get('middlewares', {})
_hdr_sec.pop(hdr_name, None)
if not _hdr_sec and 'middlewares' in config['http']:
del config['http']['middlewares']
_ledger.pop(hdr_ledger_key, None)
_ledger_changed = True
r = {'rule': rule, 'entryPoints': http_eps, 'service': service_name}
if mws:
r['middlewares'] = mws
if not no_tls:
tls_entry = {'certResolver': cert_resolver} if cert_resolver else {}
tls_main = request.form.get('tlsWildcardMain', '').strip()
tls_sans = [s.strip() for s in request.form.get('tlsWildcardSans', '').splitlines() if s.strip()]
if tls_main:
domain_entry = {'main': tls_main}
if tls_sans:
domain_entry['sans'] = tls_sans
tls_entry['domains'] = [domain_entry]
tls_opts_profile = request.form.get('tlsOptionsProfile', '').strip()
if tls_opts_profile:
tls_entry['options'] = tls_opts_profile
r['tls'] = tls_entry
if service_ref:
_merge_router(config['http']['routers'], router_name, r,
('rule', 'entryPoints', 'service', 'middlewares', 'tls'))
_orphan_tp = f"{svc_name}-transport"
_transports = config.get('http', {}).get('serversTransports', {})
if _orphan_tp in _transports:
del _transports[_orphan_tp]
if not _transports:
del config['http']['serversTransports']
else:
_be = _parse_backends_json(request.form.get('backendsJsonHttp'))
_man