jaredandClaude Opus 5.5
86aa29b378
Lint / Python (flake8) (push) Successful in 1m8s
Security / Python Security (bandit) (push) Successful in 56s
Test / Python Tests (pytest) (push) Successful in 43s
Lint / Python (flake8) (pull_request) Successful in 50s
Lint / JS (eslint) (pull_request) Successful in 30s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 55s
Test / Python Tests (pytest) (pull_request) Successful in 50s
Lint / JS (eslint) (push) Successful in 16m17s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Trace a cord: pick a panel port, press Start, and unplug the cord at the panel. The page polls a new admin-only endpoint that reads every switch port's state from UniFi (cached 2 s), shows the one port that dropped with a button to record the cable, and confirms when it's plugged back in. Several ports dropping at once, or none within 3 minutes, is explained rather than guessed at. Mismatch alerts: the monitor runs the cutsheet check every poll. Each mismatched installed run (wrong neighbor, no link, missing port, port conflict) keeps one cable_mismatch event on its switch, resolved when fixed, and opens a P4 ticket after cutsheet.mismatch_ticket_hours (default 24, 0 never). Suppressing the switch or port stops the ticket. Also fixes buttons with the hidden attribute still showing, which left a Delete button on the new-cable form. Closes #12 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1117 lines
40 KiB
Python
1117 lines
40 KiB
Python
"""Gandalf – Global Advanced Network Detection And Link Facilitator.
|
||
|
||
Flask web application serving the monitoring dashboard and suppression
|
||
management UI. Authentication via Authelia forward-auth headers.
|
||
All monitoring and alerting is handled by the separate monitor.py daemon.
|
||
"""
|
||
import hashlib
|
||
import html
|
||
import ipaddress
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from functools import wraps
|
||
|
||
from flask import Flask, jsonify, make_response, redirect, render_template, request, send_file, url_for
|
||
|
||
import cutsheet
|
||
import db
|
||
import devices
|
||
import diagnose
|
||
from monitor import PulseClient, UnifiClient
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s %(levelname)s %(name)s %(message)s',
|
||
)
|
||
logger = logging.getLogger('gandalf.web')
|
||
|
||
app = Flask(__name__)
|
||
|
||
_AVATAR_COLORS = ['lt-avatar--orange', 'lt-avatar--green', 'lt-avatar--purple', '']
|
||
|
||
|
||
@app.template_filter('avatar_color')
|
||
def avatar_color_filter(name: str) -> str:
|
||
return _AVATAR_COLORS[int(hashlib.md5(name.encode()).hexdigest(), 16) % len(_AVATAR_COLORS)] # nosec B324
|
||
|
||
_cfg = None
|
||
_cfg_lock = threading.Lock()
|
||
|
||
|
||
@app.context_processor
|
||
def inject_config():
|
||
"""Inject safe config values into all templates."""
|
||
cfg = _config()
|
||
return {
|
||
'config': {
|
||
'ticket_api': {
|
||
'web_url': cfg.get('ticket_api', {}).get('web_url', 'http://t.lotusguild.org/ticket/'),
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
# In-memory diagnostic job store { job_id: { status, result, created_at } }
|
||
_diag_jobs: dict = {}
|
||
_diag_lock = threading.Lock()
|
||
# Per-user rate-limit: { username: [epoch_float, ...] } — cleaned inside _diag_lock
|
||
_diag_rate: dict = {}
|
||
|
||
|
||
def _purge_old_jobs_loop():
|
||
"""Background thread: remove stale diagnostic jobs and mark stuck ones done."""
|
||
while True:
|
||
time.sleep(120)
|
||
cutoff = time.time() - 600
|
||
stuck_cutoff = time.time() - 300 # 5 min: job still 'running' → thread must have crashed
|
||
with _diag_lock:
|
||
stale = [jid for jid, j in _diag_jobs.items() if j.get('created_at', 0) < cutoff]
|
||
for jid in stale:
|
||
del _diag_jobs[jid]
|
||
for jid, j in list(_diag_jobs.items()):
|
||
if j['status'] == 'running' and j.get('created_at', 0) < stuck_cutoff:
|
||
j['status'] = 'done'
|
||
j['result'] = {'status': 'error', 'error': 'Diagnostic abandoned — no activity for 5 minutes.'}
|
||
logger.error(f'Diagnostic job {jid} stuck (no activity for 5 min); marked done/error')
|
||
|
||
|
||
_purge_thread = threading.Thread(target=_purge_old_jobs_loop, daemon=True)
|
||
_purge_thread.start()
|
||
|
||
|
||
def _config() -> dict:
|
||
global _cfg
|
||
if _cfg is None:
|
||
with _cfg_lock:
|
||
if _cfg is None:
|
||
with open('config.json') as f:
|
||
_cfg = json.load(f)
|
||
return _cfg
|
||
|
||
|
||
@app.after_request
|
||
def add_security_headers(response):
|
||
response.headers.setdefault('X-Content-Type-Options', 'nosniff')
|
||
response.headers.setdefault('X-Frame-Options', 'DENY')
|
||
response.headers.setdefault('Referrer-Policy', 'strict-origin-when-cross-origin')
|
||
return response
|
||
|
||
|
||
def _daemon_ok(last_check: str) -> bool:
|
||
"""Return True if monitor last checked within 20 minutes."""
|
||
if not last_check or last_check == 'Never':
|
||
return False
|
||
try:
|
||
ts = datetime.strptime(last_check, '%Y-%m-%d %H:%M:%S UTC').replace(tzinfo=timezone.utc)
|
||
return (datetime.now(timezone.utc) - ts).total_seconds() < 1200
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Auth helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _request_source() -> str:
|
||
"""The address the request came from.
|
||
|
||
Gunicorn listens on loopback behind the local nginx, which overwrites X-Real-IP
|
||
with the real peer address. A peer that isn't loopback reached gunicorn directly,
|
||
so its own address is the source and any X-Real-IP it sent is ignored.
|
||
"""
|
||
peer = request.remote_addr or ''
|
||
try:
|
||
if ipaddress.ip_address(peer).is_loopback:
|
||
return request.headers.get('X-Real-IP', '').strip()
|
||
except ValueError:
|
||
return ''
|
||
return peer
|
||
|
||
|
||
def _from_trusted_proxy() -> bool:
|
||
"""True when the request came through the Authelia-protected proxy.
|
||
|
||
Only then are the Remote-* identity headers real: anyone else on the LAN could
|
||
set them. Fails closed: with no auth.trusted_proxies configured, nobody is trusted.
|
||
"""
|
||
try:
|
||
source = ipaddress.ip_address(_request_source())
|
||
except ValueError:
|
||
return False
|
||
for entry in _config().get('auth', {}).get('trusted_proxies', []):
|
||
try:
|
||
if source in ipaddress.ip_network(entry, strict=False):
|
||
return True
|
||
except ValueError:
|
||
logger.error(f'Ignoring invalid auth.trusted_proxies entry {entry!r}')
|
||
return False
|
||
|
||
|
||
def _get_user() -> dict:
|
||
if not _from_trusted_proxy():
|
||
return {'username': '', 'name': '', 'email': '', 'groups': []}
|
||
return {
|
||
'username': request.headers.get('Remote-User', ''),
|
||
'name': request.headers.get('Remote-Name', ''),
|
||
'email': request.headers.get('Remote-Email', ''),
|
||
'groups': [
|
||
g.strip()
|
||
for g in request.headers.get('Remote-Groups', '').split(',')
|
||
if g.strip()
|
||
],
|
||
}
|
||
|
||
|
||
def require_auth(f):
|
||
@wraps(f)
|
||
def wrapper(*args, **kwargs):
|
||
user = _get_user()
|
||
if not user['username']:
|
||
if request.headers.get('Remote-User'):
|
||
logger.warning(
|
||
f'Ignored identity headers from untrusted source {_request_source() or "unknown"!r}; '
|
||
f'only auth.trusted_proxies may send them'
|
||
)
|
||
return (
|
||
'<h1>401 – Not authenticated</h1>'
|
||
'<p>Please access Gandalf through '
|
||
'<a href="https://auth.lotusguild.org">auth.lotusguild.org</a>.</p>',
|
||
401,
|
||
)
|
||
allowed = _config().get('auth', {}).get('allowed_groups', ['admin'])
|
||
if not any(g in allowed for g in user['groups']):
|
||
safe_user = html.escape(user['username'])
|
||
safe_groups = html.escape(', '.join(allowed))
|
||
return (
|
||
f'<h1>403 – Access denied</h1>'
|
||
f'<p>Your account ({safe_user}) is not in an allowed group '
|
||
f'({safe_groups}).</p>',
|
||
403,
|
||
)
|
||
return f(*args, **kwargs)
|
||
return wrapper
|
||
|
||
|
||
def require_admin(f):
|
||
"""Decorator: require require_auth AND membership in the 'admin' group."""
|
||
@wraps(f)
|
||
def wrapper(*args, **kwargs):
|
||
user = _get_user()
|
||
if 'admin' not in user.get('groups', []):
|
||
return jsonify({'error': 'Admin access required'}), 403
|
||
return f(*args, **kwargs)
|
||
return wrapper
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_PAGE_LIMIT = 200 # max events returned per request
|
||
|
||
|
||
def _annotate_suppressions(events: list, suppressions: list) -> None:
|
||
"""Annotate each event dict in-place with an is_suppressed bool.
|
||
|
||
Mirrors the suppression check order in monitor.py exactly:
|
||
interface_down → interface OR host
|
||
unifi_device_* → unifi_device
|
||
everything else → host
|
||
"""
|
||
for ev in events:
|
||
etype = ev.get('event_type', '')
|
||
name = ev.get('target_name', '')
|
||
detail = ev.get('target_detail', '') or ''
|
||
if etype == 'interface_down':
|
||
ev['is_suppressed'] = (
|
||
db.check_suppressed(suppressions, 'interface', name, detail) or
|
||
db.check_suppressed(suppressions, 'host', name)
|
||
)
|
||
elif etype == 'unifi_device_offline':
|
||
ev['is_suppressed'] = db.check_suppressed(suppressions, 'unifi_device', name, detail)
|
||
else:
|
||
ev['is_suppressed'] = db.check_suppressed(suppressions, 'host', name, detail)
|
||
|
||
|
||
def _state_json(key: str) -> dict:
|
||
"""Return a monitor_state JSON blob as a dict, or {} if missing or unreadable."""
|
||
raw = db.get_state(key)
|
||
if not raw:
|
||
return {}
|
||
try:
|
||
return json.loads(raw)
|
||
except Exception as e:
|
||
logger.error(f'Failed to parse {key} JSON: {e}')
|
||
return {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Page routes
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@app.route('/')
|
||
@require_auth
|
||
def index():
|
||
user = _get_user()
|
||
events = db.get_active_events(limit=_PAGE_LIMIT)
|
||
total_active = db.count_active_events()
|
||
summary = db.get_status_summary()
|
||
snapshot_raw = db.get_state('network_snapshot')
|
||
last_check = db.get_state('last_check', 'Never')
|
||
try:
|
||
snapshot = json.loads(snapshot_raw) if snapshot_raw else {}
|
||
except Exception as e:
|
||
logger.error(f'Failed to parse network_snapshot JSON: {e}')
|
||
snapshot = {}
|
||
suppressions = db.get_active_suppressions()
|
||
_annotate_suppressions(events, suppressions)
|
||
recent_resolved = db.get_recent_resolved(hours=24, limit=10)
|
||
return render_template(
|
||
'index.html',
|
||
user=user,
|
||
events=events,
|
||
total_active=total_active,
|
||
summary=summary,
|
||
snapshot=snapshot,
|
||
last_check=last_check,
|
||
suppressions=suppressions,
|
||
recent_resolved=recent_resolved,
|
||
daemon_ok=_daemon_ok(last_check),
|
||
)
|
||
|
||
|
||
@app.route('/links')
|
||
@require_auth
|
||
def links_page():
|
||
"""Link Debug moved to the per-device pages; keep old bookmarks working."""
|
||
return redirect(url_for('index'))
|
||
|
||
|
||
@app.route('/device/<path:name>')
|
||
@require_auth
|
||
def device_page(name: str):
|
||
return render_template('device.html', user=_get_user(), device_name=name)
|
||
|
||
|
||
@app.route('/inspector')
|
||
@require_auth
|
||
def inspector():
|
||
"""The port inspector moved to each switch's device page; keep old bookmarks working."""
|
||
return redirect(url_for('index'))
|
||
|
||
|
||
@app.route('/suppressions')
|
||
@require_auth
|
||
@require_admin
|
||
def suppressions_page():
|
||
user = _get_user()
|
||
active = db.get_active_suppressions()
|
||
history = db.get_suppression_history(limit=50)
|
||
snapshot_raw = db.get_state('network_snapshot')
|
||
try:
|
||
snapshot = json.loads(snapshot_raw) if snapshot_raw else {}
|
||
except Exception as e:
|
||
logger.error(f'Failed to parse network_snapshot JSON: {e}')
|
||
snapshot = {}
|
||
return render_template(
|
||
'suppressions.html',
|
||
user=user,
|
||
active=active,
|
||
history=history,
|
||
snapshot=snapshot,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# API routes
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.route('/api/status')
|
||
@require_auth
|
||
def api_status():
|
||
active = db.get_active_events(limit=_PAGE_LIMIT)
|
||
suppressions = db.get_active_suppressions()
|
||
_annotate_suppressions(active, suppressions)
|
||
last_check = db.get_state('last_check', 'Never')
|
||
return jsonify({
|
||
'summary': db.get_status_summary(),
|
||
'last_check': last_check,
|
||
'events': active,
|
||
'total_active': db.count_active_events(),
|
||
'daemon_ok': _daemon_ok(last_check),
|
||
})
|
||
|
||
|
||
@app.route('/api/network')
|
||
@require_auth
|
||
def api_network():
|
||
raw = db.get_state('network_snapshot')
|
||
if raw:
|
||
try:
|
||
return jsonify(json.loads(raw))
|
||
except Exception:
|
||
logger.error('Failed to parse network_snapshot JSON')
|
||
return jsonify({'hosts': {}, 'unifi': [], 'updated': None})
|
||
|
||
|
||
@app.route('/api/links')
|
||
@require_auth
|
||
def api_links():
|
||
raw = db.get_state('link_stats')
|
||
if raw:
|
||
if len(raw) > 10_000_000:
|
||
logger.error(f'link_stats exceeds 10 MB ({len(raw)} bytes); possible corruption')
|
||
return jsonify({'error': 'Invalid cached data'}), 503
|
||
try:
|
||
return jsonify(json.loads(raw))
|
||
except Exception as e:
|
||
logger.error(f'Failed to parse link_stats JSON: {e}')
|
||
return jsonify({'hosts': {}, 'unifi_switches': {}, 'updated': None})
|
||
|
||
|
||
@app.route('/api/modem')
|
||
@require_auth
|
||
def api_modem():
|
||
"""Latest WAN modem status plus optical history for the trend chart."""
|
||
try:
|
||
hours = max(1, min(int(request.args.get('hours', 168)), 2160))
|
||
except ValueError:
|
||
return jsonify({'error': 'hours must be an integer'}), 400
|
||
|
||
raw = db.get_state('modem_stats')
|
||
if not raw:
|
||
return jsonify({'configured': False})
|
||
try:
|
||
data = json.loads(raw)
|
||
except Exception as e:
|
||
logger.error(f'Failed to parse modem_stats JSON: {e}')
|
||
return jsonify({'configured': False})
|
||
|
||
# A history failure shouldn't take the current readings down with it.
|
||
try:
|
||
# Keep roughly 200 points regardless of range; never finer than the 5 min poll.
|
||
data['history'] = db.get_modem_history(hours, max(300, hours * 3600 // 200))
|
||
except Exception as e:
|
||
logger.error(f'Failed to load modem history: {e}')
|
||
data['history'] = []
|
||
data['wan_history'] = _state_json('wan_history')
|
||
data['configured'] = True
|
||
return jsonify(data)
|
||
|
||
|
||
@app.route('/api/devices')
|
||
@require_auth
|
||
def api_devices():
|
||
"""Device list for the home page search, plus cutsheet panels, jacks and cables."""
|
||
found = devices.build_index(
|
||
_state_json('network_snapshot'), _state_json('link_stats'), _state_json('modem_stats'),
|
||
)
|
||
known = {d['name'] for d in found}
|
||
try:
|
||
_ensure_cutsheet_schema()
|
||
data = db.cutsheet_load()
|
||
except Exception as e:
|
||
logger.error(f'Cutsheet unavailable for search: {e}')
|
||
data = {'items': [], 'cables': []}
|
||
for item in data['items']:
|
||
if item['name'] in known:
|
||
continue
|
||
kind = item['kind'].replace('_', ' ')
|
||
found.append({
|
||
'name': item['name'], 'kind': kind, 'status': 'unknown', 'ip': '', 'mac': '', 'model': '',
|
||
'url': f'/cutsheet#item-{item["id"]}', 'interface_count': item.get('port_count') or 0,
|
||
'search': ' '.join(filter(None, [item['name'], kind, item.get('location'), 'cutsheet'])).lower(),
|
||
})
|
||
for cable in data['cables']:
|
||
if not cable.get('label'):
|
||
continue
|
||
ends = ' '.join(filter(None, [cable['a_device'], cable['b_device']]))
|
||
found.append({
|
||
'name': cable['label'], 'kind': 'cable', 'status': 'unknown', 'ip': '', 'mac': '',
|
||
'model': f'{cable["a_device"]} ↔ {cable["b_device"] or "unterminated"}',
|
||
'url': f'/cutsheet#cable-{cable["id"]}', 'interface_count': 0,
|
||
'search': ' '.join(filter(None, [cable['label'], ends, cable.get('cable_type'),
|
||
cable.get('color'), 'cable cutsheet'])).lower(),
|
||
})
|
||
return jsonify({'devices': found})
|
||
|
||
|
||
@app.route('/api/device/<path:name>')
|
||
@require_auth
|
||
def api_device(name: str):
|
||
"""Everything known about one device, for its detail page."""
|
||
detail = devices.device_detail(
|
||
name, _state_json('network_snapshot'), _state_json('link_stats'), _state_json('modem_stats'),
|
||
)
|
||
if detail is None:
|
||
return jsonify({'error': f'No device named "{name}"'}), 404
|
||
events = db.get_events_for_target(name)
|
||
_annotate_suppressions(events, db.get_active_suppressions())
|
||
detail['events'] = events
|
||
try:
|
||
detail['cutsheet'] = _cutsheet_for_device(name)
|
||
except Exception as e:
|
||
logger.error(f'Cutsheet unavailable for {name}: {e}')
|
||
detail['cutsheet'] = []
|
||
return jsonify(detail)
|
||
|
||
|
||
@app.route('/api/events')
|
||
@require_auth
|
||
def api_events():
|
||
try:
|
||
limit = min(int(request.args.get('limit', _PAGE_LIMIT)), 1000)
|
||
offset = max(int(request.args.get('offset', 0)), 0)
|
||
except ValueError:
|
||
return jsonify({'error': 'limit and offset must be integers'}), 400
|
||
status_filter = request.args.get('status', 'active')
|
||
if status_filter not in ('active', 'resolved', 'all'):
|
||
return jsonify({'error': 'status must be active, resolved, or all'}), 400
|
||
|
||
result: dict = {}
|
||
if status_filter in ('active', 'all'):
|
||
result['active'] = db.get_active_events(limit=limit, offset=offset)
|
||
result['total_active'] = db.count_active_events()
|
||
if status_filter in ('resolved', 'all'):
|
||
result['resolved'] = db.get_recent_resolved(hours=24, limit=30)
|
||
return jsonify(result)
|
||
|
||
|
||
@app.route('/api/suppressions', methods=['GET'])
|
||
@require_auth
|
||
def api_get_suppressions():
|
||
return jsonify(db.get_active_suppressions())
|
||
|
||
|
||
@app.route('/api/suppressions', methods=['POST'])
|
||
@require_auth
|
||
@require_admin
|
||
def api_create_suppression():
|
||
user = _get_user()
|
||
data = request.get_json(silent=True) or {}
|
||
|
||
target_type = data.get('target_type', 'host')
|
||
target_name = (data.get('target_name') or '').strip()
|
||
target_detail = (data.get('target_detail') or '').strip()
|
||
reason = (data.get('reason') or '').strip()
|
||
expires_minutes = data.get('expires_minutes') # None = manual/permanent
|
||
|
||
if target_type not in ('host', 'interface', 'unifi_device', 'all'):
|
||
return jsonify({'error': 'Invalid target_type'}), 400
|
||
if target_type != 'all' and not target_name:
|
||
return jsonify({'error': 'target_name required'}), 400
|
||
if not reason:
|
||
return jsonify({'error': 'reason required'}), 400
|
||
if len(reason) > 500:
|
||
return jsonify({'error': 'reason must be 500 characters or fewer'}), 400
|
||
if len(target_name) > 255:
|
||
return jsonify({'error': 'target_name must be 255 characters or fewer'}), 400
|
||
if len(target_detail) > 255:
|
||
return jsonify({'error': 'target_detail must be 255 characters or fewer'}), 400
|
||
|
||
if expires_minutes is not None:
|
||
try:
|
||
expires_minutes = int(expires_minutes)
|
||
if expires_minutes <= 0 or expires_minutes > 43200:
|
||
return jsonify({'error': 'expires_minutes must be between 1 and 43200 (30 days)'}), 400
|
||
except (ValueError, TypeError):
|
||
return jsonify({'error': 'expires_minutes must be a valid integer'}), 400
|
||
|
||
sup_id = db.create_suppression(
|
||
target_type=target_type,
|
||
target_name=target_name,
|
||
target_detail=target_detail,
|
||
reason=reason,
|
||
suppressed_by=user['username'],
|
||
expires_minutes=expires_minutes,
|
||
)
|
||
logger.info(
|
||
f'Suppression #{sup_id} created by {user["username"]}: '
|
||
f'{target_type}/{target_name}/{target_detail} – {reason}'
|
||
)
|
||
return jsonify({'success': True, 'id': sup_id})
|
||
|
||
|
||
@app.route('/api/suppressions/<int:sup_id>', methods=['DELETE'])
|
||
@require_auth
|
||
@require_admin
|
||
def api_delete_suppression(sup_id: int):
|
||
user = _get_user()
|
||
db.deactivate_suppression(sup_id)
|
||
logger.info(f'Suppression #{sup_id} removed by {user["username"]}')
|
||
return jsonify({'success': True})
|
||
|
||
|
||
@app.route('/api/diagnose', methods=['POST'])
|
||
@require_auth
|
||
def api_diagnose_start():
|
||
"""Start a link diagnostic job. Returns {job_id}."""
|
||
data = request.get_json(silent=True) or {}
|
||
switch_name = (data.get('switch_name') or '').strip()
|
||
try:
|
||
port_idx = int(data.get('port_idx'))
|
||
except (TypeError, ValueError):
|
||
return jsonify({'error': 'port_idx must be an integer'}), 400
|
||
|
||
if not switch_name:
|
||
return jsonify({'error': 'switch_name and port_idx required'}), 400
|
||
|
||
# Look up switch + port in cached link_stats
|
||
raw = db.get_state('link_stats')
|
||
if not raw:
|
||
return jsonify({'error': 'No link_stats data available'}), 503
|
||
try:
|
||
link_data = json.loads(raw)
|
||
except Exception as e:
|
||
logger.error(f'Failed to parse link_stats JSON in /api/diagnose: {e}')
|
||
return jsonify({'error': 'Internal data error'}), 500
|
||
|
||
switches = link_data.get('unifi_switches', {})
|
||
sw = switches.get(switch_name)
|
||
if not sw:
|
||
return jsonify({'error': f'Switch "{switch_name}" not found'}), 404
|
||
|
||
# Find port by port_idx
|
||
port_data = None
|
||
for pname, pd in sw.get('ports', {}).items():
|
||
if pd.get('port_idx') == port_idx:
|
||
port_data = dict(pd)
|
||
port_data['name'] = pname
|
||
port_data['switch_name'] = switch_name
|
||
break
|
||
if not port_data:
|
||
return jsonify({'error': f'Port {port_idx} not found on switch "{switch_name}"'}), 404
|
||
|
||
# LLDP neighbor required to know which host+iface to SSH into
|
||
lldp = port_data.get('lldp')
|
||
if not lldp or not lldp.get('system_name'):
|
||
return jsonify({'error': 'No LLDP neighbor data for this port'}), 400
|
||
|
||
server_name = lldp['system_name']
|
||
if not re.fullmatch(r'[a-zA-Z0-9._-]+', server_name):
|
||
logger.error(f'Refusing diagnostic: invalid server_name from LLDP: {server_name!r}')
|
||
return jsonify({'error': 'LLDP neighbor name contains invalid characters'}), 400
|
||
lldp_port_id = lldp.get('port_id', '')
|
||
|
||
# Find matching host + interface in link_stats hosts
|
||
hosts = link_data.get('hosts', {})
|
||
server_ifaces = hosts.get(server_name)
|
||
if not server_ifaces:
|
||
return jsonify({'error': f'Host "{server_name}" not in link stats'}), 404
|
||
|
||
# Match interface by LLDP port_id (exact then fuzzy)
|
||
matched_iface = None
|
||
if lldp_port_id and lldp_port_id in server_ifaces:
|
||
matched_iface = lldp_port_id
|
||
if not matched_iface and lldp_port_id:
|
||
matched_iface = next(
|
||
(k for k in server_ifaces if lldp_port_id in k or k in lldp_port_id),
|
||
None
|
||
)
|
||
if not matched_iface:
|
||
matched_iface = next(iter(server_ifaces), None)
|
||
if not matched_iface:
|
||
return jsonify({'error': 'Cannot determine server interface'}), 400
|
||
|
||
# Resolve host IP from link_stats host data
|
||
host_ip = (server_ifaces.get(matched_iface) or {}).get('host_ip')
|
||
if not host_ip:
|
||
# Fallback: use first valid IP from LLDP mgmt IPs
|
||
for candidate in (lldp.get('mgmt_ips') or []):
|
||
try:
|
||
ipaddress.ip_address(candidate)
|
||
host_ip = candidate
|
||
break
|
||
except ValueError:
|
||
continue
|
||
if not host_ip:
|
||
return jsonify({'error': 'Cannot determine host IP for SSH'}), 400
|
||
|
||
# Validate resolved values before passing to SSH command builder
|
||
try:
|
||
ipaddress.ip_address(host_ip)
|
||
except ValueError:
|
||
logger.error(f'Refusing diagnostic: invalid host_ip "{host_ip}" for {server_name}')
|
||
return jsonify({'error': 'Resolved host IP is not a valid IP address'}), 400
|
||
if not re.fullmatch(r'[a-zA-Z0-9._-]+', matched_iface):
|
||
logger.error(f'Refusing diagnostic: invalid iface "{matched_iface}" for {server_name}')
|
||
return jsonify({'error': 'Resolved interface name contains invalid characters'}), 400
|
||
|
||
job_id = str(uuid.uuid4())
|
||
requesting_user = _get_user()['username']
|
||
now = time.time()
|
||
with _diag_lock:
|
||
# Rate limit: max 5 diagnostic jobs per user per minute; prune stale user entries
|
||
stale_users = [u for u, ts in _diag_rate.items() if not ts or max(ts) < now - 3600]
|
||
for u in stale_users:
|
||
del _diag_rate[u]
|
||
recent = [t for t in _diag_rate.get(requesting_user, []) if now - t < 60]
|
||
if len(recent) >= 5:
|
||
return jsonify({'error': 'Rate limit exceeded: max 5 diagnostics per minute'}), 429
|
||
recent.append(now)
|
||
_diag_rate[requesting_user] = recent
|
||
_diag_jobs[job_id] = {
|
||
'status': 'running', 'result': None,
|
||
'created_at': now, 'user': requesting_user,
|
||
}
|
||
|
||
def _run():
|
||
try:
|
||
cfg = _config()
|
||
pulse = PulseClient(cfg)
|
||
runner = diagnose.DiagnosticsRunner(pulse)
|
||
result = runner.run(host_ip, server_name, matched_iface, port_data)
|
||
except Exception as e:
|
||
logger.error(f'Diagnostic job {job_id} failed: {e}', exc_info=True)
|
||
result = {'status': 'error', 'error': 'Diagnostic failed; check server logs.'}
|
||
with _diag_lock:
|
||
if job_id in _diag_jobs:
|
||
_diag_jobs[job_id]['status'] = 'done'
|
||
_diag_jobs[job_id]['result'] = result
|
||
|
||
t = threading.Thread(target=_run, daemon=True)
|
||
t.start()
|
||
|
||
return jsonify({'job_id': job_id})
|
||
|
||
|
||
@app.route('/api/diagnose/<job_id>', methods=['GET'])
|
||
@require_auth
|
||
def api_diagnose_poll(job_id: str):
|
||
"""Poll a diagnostic job. Returns {status, result}."""
|
||
current_user = _get_user()['username']
|
||
with _diag_lock:
|
||
job = _diag_jobs.get(job_id)
|
||
if not job:
|
||
return jsonify({'error': 'Job not found'}), 404
|
||
if job.get('user') != current_user:
|
||
return jsonify({'error': 'Forbidden'}), 403
|
||
snapshot = {'status': job['status'], 'result': job.get('result')}
|
||
return jsonify(snapshot)
|
||
|
||
|
||
@app.route('/api/avatar')
|
||
@require_auth
|
||
def api_avatar():
|
||
"""Serve the current user's LDAP avatar photo (JPEG), cached to disk."""
|
||
username = _get_user()['username'].strip()
|
||
if not username:
|
||
return '', 404
|
||
|
||
ldap_cfg = _config().get('ldap', {})
|
||
if not ldap_cfg.get('host') or not ldap_cfg.get('bind_dn'):
|
||
logger.warning('Avatar unavailable: config.json has no ldap.host / ldap.bind_dn')
|
||
return '', 404
|
||
|
||
# Build a safe cache filename from the username (alphanumeric + - _ .)
|
||
safe_name = re.sub(r'[^a-zA-Z0-9._-]', '_', username)
|
||
cache_dir = os.path.abspath(
|
||
ldap_cfg.get('cache_dir', os.path.join(tempfile.gettempdir(), 'gandalf_avatars'))
|
||
)
|
||
os.makedirs(cache_dir, exist_ok=True)
|
||
cache_file = os.path.abspath(os.path.join(cache_dir, f'user_{safe_name}.jpg'))
|
||
sentinel = os.path.abspath(os.path.join(cache_dir, f'user_{safe_name}.none'))
|
||
# Guard against path escape (shouldn't happen with sanitised safe_name, but be explicit)
|
||
if not cache_file.startswith(cache_dir + os.sep) or not sentinel.startswith(cache_dir + os.sep):
|
||
logger.error(f'Avatar path escape detected for user {username!r}')
|
||
return '', 404
|
||
try:
|
||
cache_ttl = int(ldap_cfg.get('cache_ttl', 3600))
|
||
except (ValueError, TypeError):
|
||
logger.warning('Invalid cache_ttl in ldap config; using default 3600')
|
||
cache_ttl = 3600
|
||
|
||
now = time.time()
|
||
|
||
# Serve cached image if fresh
|
||
if os.path.exists(cache_file) and now - os.path.getmtime(cache_file) < cache_ttl:
|
||
return send_file(cache_file, mimetype='image/jpeg',
|
||
max_age=cache_ttl, conditional=True)
|
||
|
||
# Skip LDAP if we already know this user has no avatar
|
||
try:
|
||
if os.path.exists(sentinel) and now - os.path.getmtime(sentinel) < cache_ttl:
|
||
logger.info(f'Avatar for {username!r}: cached "no photo" marker is still valid')
|
||
return '', 404
|
||
except OSError:
|
||
pass
|
||
|
||
# Query lldap
|
||
bind_pw = ldap_cfg.get('bind_pw', '')
|
||
if not bind_pw:
|
||
logger.error('LDAP bind_pw not configured — avatar lookup disabled')
|
||
return '', 404
|
||
|
||
avatar_data = None
|
||
conn = None
|
||
try:
|
||
import ldap3
|
||
server = ldap3.Server(ldap_cfg['host'], port=int(ldap_cfg.get('port', 3890)))
|
||
conn = ldap3.Connection(server,
|
||
user=ldap_cfg['bind_dn'],
|
||
password=bind_pw,
|
||
auto_bind=True, receive_timeout=5)
|
||
safe_uid = ldap3.utils.conv.escape_filter_chars(username)
|
||
conn.search(ldap_cfg.get('user_base', 'ou=people,dc=example,dc=com'),
|
||
f'(uid={safe_uid})', attributes=['avatar'])
|
||
if conn.entries and conn.entries[0]['avatar'].value:
|
||
avatar_data = conn.entries[0]['avatar'].value
|
||
except ImportError:
|
||
logger.error('ldap3 not installed — run: pip install ldap3')
|
||
return '', 404
|
||
except Exception as e:
|
||
logger.error(f'LDAP avatar lookup failed for {username}: {e}')
|
||
return '', 404
|
||
finally:
|
||
if conn is not None:
|
||
try:
|
||
conn.unbind()
|
||
except Exception:
|
||
pass
|
||
|
||
if not avatar_data or len(avatar_data) < 100:
|
||
logger.warning(f'Avatar for {username!r}: LDAP has no photo (or no user with that uid)')
|
||
with open(sentinel, 'w'):
|
||
pass
|
||
return '', 404
|
||
|
||
# Validate JPEG magic bytes (FF D8 FF)
|
||
if isinstance(avatar_data, str):
|
||
avatar_data = avatar_data.encode('latin-1')
|
||
if avatar_data[:3] != b'\xFF\xD8\xFF':
|
||
logger.warning(f'Non-JPEG avatar data for {username}')
|
||
with open(sentinel, 'w'):
|
||
pass
|
||
return '', 404
|
||
|
||
with open(cache_file, 'wb') as f:
|
||
f.write(avatar_data)
|
||
if os.path.exists(sentinel):
|
||
os.unlink(sentinel)
|
||
|
||
resp = make_response(avatar_data)
|
||
resp.headers['Content-Type'] = 'image/jpeg'
|
||
resp.headers['Cache-Control'] = f'private, max-age={cache_ttl}'
|
||
return resp
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Cutsheets
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_cutsheet_schema_ready = False
|
||
_cutsheet_schema_lock = threading.Lock()
|
||
|
||
|
||
def _ensure_cutsheet_schema() -> None:
|
||
"""Create the cutsheet tables on first use, so deploys need no migration step."""
|
||
global _cutsheet_schema_ready
|
||
if _cutsheet_schema_ready:
|
||
return
|
||
with _cutsheet_schema_lock:
|
||
if not _cutsheet_schema_ready:
|
||
db.ensure_cutsheet_schema()
|
||
_cutsheet_schema_ready = True
|
||
|
||
|
||
def _node_json(node) -> dict:
|
||
if node is None:
|
||
return {'device': '', 'port': '', 'side': '', 'text': 'unterminated'}
|
||
return {'device': node[0], 'port': node[1], 'side': node[2], 'text': cutsheet.describe_node(node)}
|
||
|
||
|
||
def _cutsheet_catalog(link_stats: dict, items: list) -> list:
|
||
"""Devices and their ports, for the cable form's suggestions."""
|
||
catalog = []
|
||
for name, sw in sorted((link_stats.get('unifi_switches') or {}).items()):
|
||
ports = sorted((p for p in (sw.get('ports') or {}).items() if p[1].get('port_idx') is not None),
|
||
key=lambda kv: kv[1]['port_idx'])
|
||
catalog.append({'name': name, 'kind': 'switch', 'ports': [
|
||
{'value': str(p['port_idx']), 'text': f'{p["port_idx"]} · {pname}'} for pname, p in ports]})
|
||
for name, ifaces in sorted((link_stats.get('hosts') or {}).items()):
|
||
catalog.append({'name': name, 'kind': 'host',
|
||
'ports': [{'value': i, 'text': i} for i in sorted(ifaces)]})
|
||
known = {c['name'] for c in catalog}
|
||
for item in items:
|
||
if item['name'] in known:
|
||
continue
|
||
ports = [{'value': str(n), 'text': str(n)} for n in range(1, (item.get('port_count') or 0) + 1)]
|
||
catalog.append({'name': item['name'], 'kind': item['kind'], 'ports': ports})
|
||
return catalog
|
||
|
||
|
||
def _cutsheet_view() -> dict:
|
||
"""Every cutsheet record with its traced path and plan-vs-actual verdict."""
|
||
_ensure_cutsheet_schema()
|
||
data = db.cutsheet_load()
|
||
link_stats = _state_json('link_stats')
|
||
switches = link_stats.get('unifi_switches') or {}
|
||
items = {i['name']: i for i in data['items']}
|
||
results = cutsheet.evaluate(data['cables'], items, switches)
|
||
for cable in data['cables']:
|
||
r = results[cable['id']]
|
||
cable['verdict'] = r['verdict']
|
||
cable['detail'] = r['detail']
|
||
cable['problem'] = r['verdict'] in cutsheet.PROBLEM_VERDICTS
|
||
cable['path'] = [
|
||
{'cable_id': h} if isinstance(h, int) else _node_json(h) for h in r['path']['hops']
|
||
]
|
||
cable['a_text'] = cutsheet.describe_node(cutsheet._node(cable, 'a'))
|
||
cable['b_text'] = cutsheet.describe_node(cutsheet._node(cable, 'b'))
|
||
data['suggestions'] = cutsheet.undocumented(data['cables'], switches)
|
||
data['labels'] = cutsheet.labels_to_make(data['cables'], items, data['port_labels'])
|
||
data['next_label'] = cutsheet.next_label(data['cables'])
|
||
data['catalog'] = _cutsheet_catalog(link_stats, data['items'])
|
||
data['verdicts'] = cutsheet.VERDICTS
|
||
data['options'] = {'kinds': cutsheet.ITEM_KINDS, 'passive_kinds': cutsheet.PASSIVE_KINDS,
|
||
'statuses': cutsheet.STATUSES, 'cable_types': cutsheet.CABLE_TYPES}
|
||
data['updated'] = link_stats.get('updated')
|
||
return data
|
||
|
||
|
||
def _cutsheet_for_device(name: str) -> list:
|
||
"""The cables that end on one device, for its port panel."""
|
||
view = _cutsheet_view()
|
||
out = []
|
||
for cable in view['cables']:
|
||
for end in ('a', 'b'):
|
||
if cable[f'{end}_device'] == name and cable['status'] != 'retired':
|
||
out.append({
|
||
'port': cable[f'{end}_port'], 'side': cable[f'{end}_side'], 'id': cable['id'],
|
||
'label': cable['label'], 'status': cable['status'], 'verdict': cable['verdict'],
|
||
'detail': cable['detail'], 'problem': cable['problem'], 'path': cable['path'],
|
||
})
|
||
return out
|
||
|
||
|
||
def _json_body() -> dict:
|
||
"""The request's JSON object. Requiring a JSON content type also keeps another
|
||
site's plain form post from reaching these write endpoints (no CORS preflight)."""
|
||
if not request.is_json:
|
||
raise cutsheet.ValidationError('Send the request as JSON')
|
||
data = request.get_json(silent=True)
|
||
if not isinstance(data, dict):
|
||
raise cutsheet.ValidationError('Request body must be a JSON object')
|
||
return data
|
||
|
||
|
||
def _cutsheet_write(fn):
|
||
"""Turn validation and conflict errors from a write into 400 and 409 responses."""
|
||
@wraps(fn)
|
||
def wrapper(*args, **kwargs):
|
||
try:
|
||
_ensure_cutsheet_schema()
|
||
return fn(*args, **kwargs)
|
||
except cutsheet.ValidationError as e:
|
||
return jsonify({'error': str(e)}), 400
|
||
except db.CutsheetConflict as e:
|
||
return jsonify({'error': str(e)}), 409
|
||
return wrapper
|
||
|
||
|
||
def _cutsheet_items() -> dict:
|
||
return {i['name']: i for i in db.cutsheet_load()['items']}
|
||
|
||
|
||
@app.route('/cutsheet')
|
||
@require_auth
|
||
def cutsheet_page():
|
||
user = _get_user()
|
||
return render_template('cutsheet.html', user=user, can_edit='admin' in user['groups'])
|
||
|
||
|
||
@app.route('/api/cutsheet')
|
||
@require_auth
|
||
def api_cutsheet():
|
||
view = _cutsheet_view()
|
||
view['history'] = db.cutsheet_history(limit=100)
|
||
return jsonify(view)
|
||
|
||
|
||
@app.route('/api/cutsheet/cables', methods=['POST'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_create_cable():
|
||
cable = cutsheet.validate_cable(_json_body(), _cutsheet_items())
|
||
cable_id = db.cutsheet_create_cable(cable, _get_user()['username'])
|
||
return jsonify({'success': True, 'id': cable_id})
|
||
|
||
|
||
@app.route('/api/cutsheet/cables/<int:cable_id>', methods=['PUT'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_update_cable(cable_id: int):
|
||
cable = cutsheet.validate_cable(_json_body(), _cutsheet_items())
|
||
if not db.cutsheet_update_cable(cable_id, cable, _get_user()['username']):
|
||
return jsonify({'error': 'No such cable'}), 404
|
||
return jsonify({'success': True})
|
||
|
||
|
||
@app.route('/api/cutsheet/cables/<int:cable_id>', methods=['DELETE'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_delete_cable(cable_id: int):
|
||
if not db.cutsheet_delete_cable(cable_id, _get_user()['username']):
|
||
return jsonify({'error': 'No such cable'}), 404
|
||
return jsonify({'success': True})
|
||
|
||
|
||
@app.route('/api/cutsheet/items', methods=['POST'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_create_item():
|
||
item = cutsheet.validate_item(_json_body())
|
||
item_id = db.cutsheet_create_item(item, _get_user()['username'])
|
||
return jsonify({'success': True, 'id': item_id})
|
||
|
||
|
||
@app.route('/api/cutsheet/items/<int:item_id>', methods=['PUT'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_update_item(item_id: int):
|
||
item = cutsheet.validate_item(_json_body())
|
||
if not db.cutsheet_update_item(item_id, item, _get_user()['username']):
|
||
return jsonify({'error': 'No such item'}), 404
|
||
return jsonify({'success': True})
|
||
|
||
|
||
@app.route('/api/cutsheet/items/<int:item_id>', methods=['DELETE'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_delete_item(item_id: int):
|
||
if not db.cutsheet_delete_item(item_id, _get_user()['username']):
|
||
return jsonify({'error': 'No such item'}), 404
|
||
return jsonify({'success': True})
|
||
|
||
|
||
@app.route('/api/cutsheet/items/<int:item_id>/ports/<port>', methods=['PUT'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_port_label(item_id: int, port: str):
|
||
data = _json_body()
|
||
label = str(data.get('label') or '').strip()
|
||
if len(label) > 64:
|
||
raise cutsheet.ValidationError('Port label must be 64 characters or fewer')
|
||
if len(port) > cutsheet.PORT_MAX:
|
||
raise cutsheet.ValidationError('Port is too long')
|
||
if not db.cutsheet_set_port_label(item_id, port, label, bool(data.get('applied')), _get_user()['username']):
|
||
return jsonify({'error': 'No such item'}), 404
|
||
return jsonify({'success': True})
|
||
|
||
|
||
_live_ports_cache: dict = {'at': 0.0, 'data': None, 'client': None}
|
||
_live_ports_lock = threading.Lock()
|
||
_LIVE_PORTS_TTL = 2.0
|
||
|
||
|
||
@app.route('/api/cutsheet/live-ports')
|
||
@require_auth
|
||
@require_admin
|
||
def api_cutsheet_live_ports():
|
||
"""Current up/down state of every switch port, read from UniFi now.
|
||
|
||
Trace mode polls this while someone unplugs a cord, to see which port dropped.
|
||
The monitor's cached state is minutes old, so it can't be used for this.
|
||
"""
|
||
with _live_ports_lock:
|
||
now = time.time()
|
||
if _live_ports_cache['data'] is None or now - _live_ports_cache['at'] > _LIVE_PORTS_TTL:
|
||
if _live_ports_cache['client'] is None:
|
||
_live_ports_cache['client'] = UnifiClient(_config()['unifi'])
|
||
raw = _live_ports_cache['client'].get_switch_ports()
|
||
if raw is None:
|
||
return jsonify({'error': 'The UniFi controller did not answer'}), 502
|
||
_live_ports_cache['data'] = {
|
||
sw: {str(p['port_idx']): {'up': bool(p.get('up')), 'name': name}
|
||
for name, p in (d.get('ports') or {}).items() if p.get('port_idx') is not None}
|
||
for sw, d in raw.items()
|
||
}
|
||
_live_ports_cache['at'] = now
|
||
return jsonify({'switches': _live_ports_cache['data'], 'at': _live_ports_cache['at']})
|
||
|
||
|
||
@app.route('/api/cutsheet/cables.csv')
|
||
@require_auth
|
||
def api_cutsheet_export():
|
||
_ensure_cutsheet_schema()
|
||
resp = make_response(cutsheet.cables_to_csv(db.cutsheet_load()['cables']))
|
||
resp.headers['Content-Type'] = 'text/csv; charset=utf-8'
|
||
resp.headers['Content-Disposition'] = 'attachment; filename="gandalf-cutsheet.csv"'
|
||
return resp
|
||
|
||
|
||
@app.route('/api/cutsheet/import', methods=['POST'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_import():
|
||
data = _json_body()
|
||
text = data.get('csv')
|
||
if not isinstance(text, str) or not text.strip():
|
||
raise cutsheet.ValidationError('Paste or choose a CSV file')
|
||
if len(text) > 1_000_000:
|
||
raise cutsheet.ValidationError('CSV is larger than 1 MB')
|
||
cables, errors = cutsheet.parse_csv(text, _cutsheet_items())
|
||
if errors:
|
||
return jsonify({'error': 'Nothing was imported', 'errors': errors[:50]}), 400
|
||
if data.get('dry_run'):
|
||
return jsonify({'success': True, 'dry_run': True, 'rows': len(cables)})
|
||
result = db.cutsheet_import_cables(cables, _get_user()['username'])
|
||
return jsonify(dict(result, success=True))
|
||
|
||
|
||
@app.route('/health')
|
||
def health():
|
||
"""Health check endpoint (no auth). Checks DB and monitor freshness."""
|
||
checks = {}
|
||
overall = 'ok'
|
||
|
||
# DB connectivity
|
||
try:
|
||
db.get_state('last_check')
|
||
checks['db'] = 'ok'
|
||
except Exception as e:
|
||
logger.error(f'Health check db error: {e}')
|
||
checks['db'] = 'error'
|
||
overall = 'degraded'
|
||
|
||
# Monitor freshness: fail if last_check is older than 20 minutes
|
||
try:
|
||
last_check = db.get_state('last_check', '')
|
||
if last_check:
|
||
ts = datetime.strptime(last_check, '%Y-%m-%d %H:%M:%S UTC').replace(tzinfo=timezone.utc)
|
||
age_s = (datetime.now(timezone.utc) - ts).total_seconds()
|
||
if age_s > 1200:
|
||
checks['monitor'] = 'stale'
|
||
overall = 'degraded'
|
||
else:
|
||
checks['monitor'] = 'ok'
|
||
else:
|
||
checks['monitor'] = 'no data yet'
|
||
except Exception as e:
|
||
logger.error(f'Health check monitor error: {e}')
|
||
checks['monitor'] = 'error'
|
||
overall = 'degraded'
|
||
|
||
status_code = 200 if overall == 'ok' else 503
|
||
return jsonify({'status': overall, 'service': 'gandalf', 'checks': checks}), status_code
|
||
|
||
|
||
if __name__ == '__main__':
|
||
app.run(debug=True, host='0.0.0.0', port=5000) # nosec B201 B104 — dev runner only; production uses gunicorn
|