jaredandClaude Opus 5.5
c50e1f8322
Lint / JS (eslint) (push) Successful in 7s
Lint / Python (flake8) (push) Successful in 49s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 51s
Lint / Python (flake8) (pull_request) Successful in 55s
Test / Python Tests (pytest) (push) Successful in 1m4s
Lint / JS (eslint) (pull_request) Successful in 7s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 52s
Test / Python Tests (pytest) (pull_request) Successful in 55s
Items are mounted from the front, the rear, or full depth. On a 4-post rack a front and a rear item can share a unit; a full-depth one takes both sides, and the rear view shows its back. Overlaps are checked per side. Shelf contents take the shelf's side. A Front / Rear switch flips the drawing. Racks are saved in Gandalf (new cutsheet_racks table) and edited on the rack page by admins: name, height, notes, add and delete. Renaming moves the rack's items; saving the first rack carries over the items of the config default it replaces. Config racks remain the fallback. Faceplates are stretched to the full width only when the drawing is close to the slot's proportions, so the PDU is not squashed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1418 lines
53 KiB
Python
1418 lines
53 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 typing import Optional
|
||
from urllib.parse import quote, urlparse
|
||
|
||
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 _monitor_stale_after() -> int:
|
||
"""Seconds without a completed check before the monitor counts as stalled.
|
||
|
||
Five missed polls, and never under 10 minutes: one slow cycle (an SSH timeout
|
||
to a dead host) shouldn't trip it, but a hung monitor is caught in minutes.
|
||
"""
|
||
try:
|
||
poll = int(_config().get('monitor', {}).get('poll_interval', 120))
|
||
except (TypeError, ValueError):
|
||
poll = 120
|
||
return max(600, poll * 5)
|
||
|
||
|
||
def _monitor_age(last_check: str) -> Optional[float]:
|
||
"""Seconds since the monitor's last check, or None if it never ran or can't be read."""
|
||
if not last_check or last_check == 'Never':
|
||
return None
|
||
try:
|
||
ts = datetime.strptime(last_check, '%Y-%m-%d %H:%M:%S UTC').replace(tzinfo=timezone.utc)
|
||
except ValueError:
|
||
return None
|
||
return (datetime.now(timezone.utc) - ts).total_seconds()
|
||
|
||
|
||
def _daemon_ok(last_check: str) -> bool:
|
||
"""True if the monitor checked recently enough (see _monitor_stale_after)."""
|
||
age = _monitor_age(last_check)
|
||
return age is not None and age < _monitor_stale_after()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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):
|
||
user = _get_user()
|
||
return render_template('device.html', user=user, device_name=name, can_edit='admin' in user['groups'])
|
||
|
||
|
||
@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('/events')
|
||
@require_auth
|
||
def events_page():
|
||
return render_template('events.html', user=_get_user())
|
||
|
||
|
||
_HISTORY_RANGES = {'24h': 24, '7d': 24 * 7, '30d': 24 * 30, '90d': 24 * 90}
|
||
_HISTORY_PAGE = 100
|
||
|
||
|
||
@app.route('/api/events/history')
|
||
@require_auth
|
||
def api_events_history():
|
||
"""Every alert open at some point in a time window, with summaries.
|
||
|
||
Query: range (24h|7d|30d|90d), type, target (substring of name or detail),
|
||
severity, status (all|active|resolved), offset.
|
||
"""
|
||
args = request.args
|
||
hours = _HISTORY_RANGES.get(args.get('range', '7d'))
|
||
if hours is None:
|
||
return jsonify({'error': f'range must be one of {", ".join(_HISTORY_RANGES)}'}), 400
|
||
severity = args.get('severity', '')
|
||
if severity not in ('', 'critical', 'warning', 'info'):
|
||
return jsonify({'error': 'severity must be critical, warning or info'}), 400
|
||
status = args.get('status', 'all')
|
||
if status not in ('all', 'active', 'resolved'):
|
||
return jsonify({'error': 'status must be all, active or resolved'}), 400
|
||
try:
|
||
offset = max(int(args.get('offset', 0)), 0)
|
||
except ValueError:
|
||
return jsonify({'error': 'offset must be an integer'}), 400
|
||
event_type = args.get('type', '')[:60]
|
||
target = args.get('target', '').strip()[:255]
|
||
|
||
result = db.get_events_history(hours, event_type, target, severity, status,
|
||
limit=_HISTORY_PAGE, offset=offset)
|
||
result['flappers'] = db.get_flappers(hours, event_type, target, severity)
|
||
result['types'] = db.get_event_types()
|
||
result['offset'] = offset
|
||
result['page_size'] = _HISTORY_PAGE
|
||
return jsonify(result)
|
||
|
||
|
||
@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'] = []
|
||
detail['inventory'] = _inventory_for(name, detail.get('model') or '')
|
||
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, items)
|
||
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['racks'] = [r['name'] for r in _racks(data['items'])]
|
||
snapshot = _state_json('network_snapshot')
|
||
data['devices'] = sorted(set(switches) | set(snapshot.get('hosts') or {})
|
||
| {u['name'] for u in snapshot.get('unifi') or [] if u.get('name')}, key=str.lower)
|
||
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()
|
||
# Cables may name the device or a cutsheet item linked to it
|
||
names = {name} | {i['name'] for i in view['items'] if i.get('device') == name}
|
||
out = []
|
||
for cable in view['cables']:
|
||
for end in ('a', 'b'):
|
||
if cable[f'{end}_device'] in names 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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Inventory (HomeBox) links
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_INVENTORY_DEFAULT = 'https://inventory.lotusguild.org'
|
||
|
||
|
||
def _inventory_base() -> str:
|
||
return ((_config().get('inventory') or {}).get('base_url') or _INVENTORY_DEFAULT).rstrip('/')
|
||
|
||
|
||
def _inventory_for(name: str, model: str) -> dict:
|
||
"""The device's inventory record link, or a search to find it."""
|
||
base = _inventory_base()
|
||
template = (_config().get('inventory') or {}).get('search_url') or '{base}/items?q={q}'
|
||
try:
|
||
_ensure_cutsheet_schema()
|
||
url = db.get_device_link(name)
|
||
except Exception as e:
|
||
logger.error(f'Inventory link unavailable for {name}: {e}')
|
||
url = ''
|
||
return {'url': url, 'search_url': template.format(base=base, q=quote(model or name, safe=''))}
|
||
|
||
|
||
def _validate_inventory_url(value) -> str:
|
||
"""Only links into the configured inventory site: this URL is shown as a link to every viewer."""
|
||
url = str(value or '').strip()
|
||
if not url:
|
||
return ''
|
||
if len(url) > 512:
|
||
raise cutsheet.ValidationError('Inventory link is too long')
|
||
base = urlparse(_inventory_base())
|
||
got = urlparse(url)
|
||
if got.scheme != 'https' or got.netloc.lower() != base.netloc.lower() or got.username or got.password:
|
||
raise cutsheet.ValidationError(f'Inventory link must start with {base.scheme}://{base.netloc}/')
|
||
return url
|
||
|
||
|
||
@app.route('/api/device/<path:name>/inventory', methods=['PUT'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_device_inventory(name: str):
|
||
"""Set or clear (empty url) the device's inventory link."""
|
||
if len(name) > 255:
|
||
raise cutsheet.ValidationError('Device name is too long')
|
||
url = _validate_inventory_url(_json_body().get('url'))
|
||
db.set_device_link(name, url, _get_user()['username'])
|
||
logger.info(f'{_get_user()["username"]} set inventory link for {name}: {url or "(removed)"}')
|
||
return jsonify({'success': True, 'url': url})
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Rack elevation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.route('/rack')
|
||
@require_auth
|
||
def rack_page():
|
||
user = _get_user()
|
||
return render_template('rack.html', user=user, can_edit='admin' in user['groups'])
|
||
|
||
|
||
def _racks(items: list) -> list:
|
||
"""Every rack, in order: those saved on the rack page, then any other name an item uses.
|
||
|
||
With none saved, racks come from config: "racks": [{"name": "Main", "units": 42}, ...], or
|
||
one rack named rack.name (default "Rack 1") with rack.units (default 42). Those have no id
|
||
until saved. An item with no rack belongs to the first.
|
||
"""
|
||
def units(value, default=42):
|
||
try:
|
||
return max(1, min(int(value), 60))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
cfg = _config()
|
||
single = cfg.get('rack') or {}
|
||
default_units = units(single.get('units', 42))
|
||
racks = [{'id': r['id'], 'name': r['name'], 'units': r['units'], 'notes': r.get('notes') or ''}
|
||
for r in db.cutsheet_racks()]
|
||
if not racks:
|
||
racks = [{'id': None, 'name': str(r.get('name') or '').strip(), 'notes': '',
|
||
'units': units(r.get('units', default_units), default_units)}
|
||
for r in cfg.get('racks') or [] if isinstance(r, dict) and str(r.get('name') or '').strip()]
|
||
if not racks:
|
||
racks = [{'id': None, 'name': str(single.get('name') or 'Rack 1'), 'units': default_units, 'notes': ''}]
|
||
named = {r['name'] for r in racks}
|
||
for item in items:
|
||
if item.get('rack') and item['rack'] not in named:
|
||
named.add(item['rack'])
|
||
racks.append({'id': None, 'name': item['rack'], 'units': default_units, 'notes': ''})
|
||
return racks
|
||
|
||
|
||
def _device_status(name: str, snapshot: dict) -> str:
|
||
host = (snapshot.get('hosts') or {}).get(name)
|
||
if host:
|
||
return host.get('status') or 'unknown'
|
||
for d in snapshot.get('unifi') or []:
|
||
if d.get('name') == name:
|
||
return 'up' if d.get('connected') else 'down'
|
||
return 'unknown'
|
||
|
||
|
||
@app.route('/api/rack')
|
||
@require_auth
|
||
def api_rack():
|
||
"""One rack face: placed cutsheet items, live switch ports, and each port's cutsheet state.
|
||
|
||
Query: rack (name; default the first rack), face (front|rear).
|
||
"""
|
||
face = request.args.get('face', 'front')
|
||
if face not in ('front', 'rear'):
|
||
return jsonify({'error': 'face must be front or rear'}), 400
|
||
view = _cutsheet_view()
|
||
link_stats = _state_json('link_stats')
|
||
switches = link_stats.get('unifi_switches') or {}
|
||
unifi_info = link_stats.get('unifi_devices') or {}
|
||
snapshot = _state_json('network_snapshot')
|
||
online = {d['name']: d.get('connected', True) for d in snapshot.get('unifi') or []}
|
||
known = set(switches) | set(online) | set(snapshot.get('hosts') or {})
|
||
|
||
racks = _racks(view['items'])
|
||
default = racks[0]['name']
|
||
current = next((r for r in racks if r['name'] == request.args.get('rack', default)), None)
|
||
if current is None:
|
||
return jsonify({'error': 'No such rack'}), 404
|
||
|
||
def rack_of(item):
|
||
return item.get('rack') or default
|
||
|
||
for r in racks:
|
||
racked = [i for i in view['items'] if not i.get('shelf') and i.get('rack_u') and rack_of(i) == r['name']]
|
||
r['count'] = len(racked)
|
||
r['rear'] = sum(1 for i in racked if i.get('mount') in ('rear', 'full'))
|
||
for cable in view['cables']:
|
||
cable['run'] = [h['cable_id'] for h in cable['path'] if 'cable_id' in h]
|
||
marks = cutsheet.port_marks(view['cables'])
|
||
labels: dict = {}
|
||
for pl in view['port_labels']:
|
||
labels.setdefault(pl['item_name'], {})[str(pl['port'])] = pl['label']
|
||
|
||
def dev(item):
|
||
# An item can be named after the product ("UDM-Pro") and linked to the live device
|
||
return item.get('device') or item['name']
|
||
|
||
def url(item):
|
||
return f'/device/{quote(dev(item), safe="")}' if dev(item) in known else f'/cutsheet#item-{item["id"]}'
|
||
|
||
on_shelf: dict = {}
|
||
for item in view['items']:
|
||
if item.get('shelf'):
|
||
on_shelf.setdefault(item['shelf'], []).append(item)
|
||
|
||
here = [i for i in view['items'] if not i.get('shelf') and rack_of(i) == current['name']]
|
||
layout = cutsheet.rack_layout(here, current['units'], face)
|
||
out = []
|
||
for item in layout['placed']:
|
||
name, device = item['name'], dev(item)
|
||
entry = {k: item.get(k) for k in ('id', 'name', 'kind', 'port_count', 'rack_u', 'rack_height',
|
||
'rack_top', 'location', 'notes')}
|
||
entry['device'] = device
|
||
entry['linked'] = device in known
|
||
entry['mount'] = item.get('mount') or 'front'
|
||
entry['through'] = item.get('through', False)
|
||
entry['url'] = url(item)
|
||
entry['status'] = _device_status(device, snapshot) if device in known else ''
|
||
# Cables may name either the item or the device it is linked to
|
||
entry['ports'] = {port: m for (d, port), m in marks.items() if d in (name, device)}
|
||
entry['labels'] = labels.get(name, {})
|
||
model = next((u.get('model') for u in snapshot.get('unifi') or [] if u.get('name') == device), '')
|
||
if device in switches or model:
|
||
sw = switches.get(device) or {}
|
||
entry['switch'] = {'model': sw.get('model') or model or '', 'ports': sw.get('ports') or {},
|
||
'outlets': (unifi_info.get(device) or {}).get('outlets') or [],
|
||
'online': online.get(device, True)}
|
||
if item['kind'] == 'shelf':
|
||
entry['contents'] = [
|
||
{'id': c['id'], 'name': c['name'], 'kind': c['kind'], 'url': url(c),
|
||
'status': _device_status(dev(c), snapshot) if dev(c) in known else '',
|
||
'model': next((u.get('model') for u in snapshot.get('unifi') or [] if u.get('name') == dev(c)), '')
|
||
or '', 'notes': c.get('notes') or ''}
|
||
for c in sorted(on_shelf.get(name, []), key=lambda c: c['name'].lower())
|
||
]
|
||
out.append(entry)
|
||
return jsonify({
|
||
'racks': racks, 'rack': current['name'], 'rack_id': current['id'], 'units': current['units'],
|
||
'notes': current['notes'], 'face': face, 'items': out,
|
||
'overlaps': layout['overlaps'], 'outside': layout['outside'],
|
||
'unplaced': sorted(i['name'] for i in here if not i.get('rack_u')),
|
||
'updated': link_stats.get('updated'),
|
||
})
|
||
|
||
|
||
@app.route('/api/cutsheet/racks', methods=['POST'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_create_rack():
|
||
"""Add a rack. `previous` is the name of the default rack this replaces, when the first is saved."""
|
||
body = _json_body()
|
||
rack = cutsheet.validate_rack(body)
|
||
previous = str(body.get('previous') or '')[:100] if not db.cutsheet_racks() else ''
|
||
rack_id = db.cutsheet_save_rack(None, rack, _get_user()['username'], previous=previous)
|
||
return jsonify({'success': True, 'id': rack_id})
|
||
|
||
|
||
@app.route('/api/cutsheet/racks/<int:rack_id>', methods=['PUT'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_update_rack(rack_id: int):
|
||
rack = cutsheet.validate_rack(_json_body())
|
||
db.cutsheet_save_rack(rack_id, rack, _get_user()['username'])
|
||
return jsonify({'success': True})
|
||
|
||
|
||
@app.route('/api/cutsheet/racks/<int:rack_id>', methods=['DELETE'])
|
||
@require_auth
|
||
@require_admin
|
||
@_cutsheet_write
|
||
def api_cutsheet_delete_rack(rack_id: int):
|
||
if not db.cutsheet_delete_rack(rack_id, _get_user()['username']):
|
||
return jsonify({'error': 'No such rack'}), 404
|
||
return jsonify({'success': True})
|
||
|
||
|
||
@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(), _cutsheet_items())
|
||
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(), _cutsheet_items())
|
||
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: a hung or crashed monitor stops updating last_check
|
||
try:
|
||
last_check = db.get_state('last_check', '')
|
||
age_s = _monitor_age(last_check)
|
||
if not last_check:
|
||
checks['monitor'] = 'no data yet'
|
||
elif age_s is None or age_s > _monitor_stale_after():
|
||
checks['monitor'] = 'stale'
|
||
checks['monitor_age_s'] = int(age_s) if age_s is not None else None
|
||
overall = 'degraded'
|
||
else:
|
||
checks['monitor'] = 'ok'
|
||
checks['monitor_age_s'] = int(age_s)
|
||
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
|