253 lines
7.4 KiB
Python
253 lines
7.4 KiB
Python
"""
|
|
Web Dashboard for A/D Infrastructure Control
|
|
Flask-based dashboard to monitor services, attacks, and alerts
|
|
"""
|
|
import os
|
|
import asyncio
|
|
from datetime import datetime
|
|
from flask import Flask, render_template, jsonify, request, redirect, url_for, session
|
|
import aiohttp
|
|
from functools import wraps
|
|
|
|
# Configuration
|
|
SECRET_TOKEN = os.getenv("SECRET_TOKEN", "change-me-in-production")
|
|
WEB_PASSWORD = os.getenv("WEB_PASSWORD", "admin123")
|
|
CONTROLLER_API = os.getenv("CONTROLLER_API", "http://controller:8001")
|
|
SCOREBOARD_API = os.getenv("SCOREBOARD_API", "http://scoreboard-injector:8002")
|
|
TELEGRAM_API = os.getenv("TELEGRAM_API", "http://tg-bot:8003")
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.getenv("FLASK_SECRET_KEY", "change-me-in-production-flask-secret")
|
|
|
|
# Auth decorator
|
|
def login_required(f):
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if not session.get('logged_in'):
|
|
return redirect(url_for('login'))
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
# API call helper
|
|
async def api_call(url: str, method: str = "GET", data: dict = {}):
|
|
"""Make API call to internal services"""
|
|
headers = {"Authorization": f"Bearer {SECRET_TOKEN}"}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
if method == "GET":
|
|
async with session.get(url, headers=headers) as resp:
|
|
if resp.status == 200:
|
|
return await resp.json()
|
|
return {"error": f"Status {resp.status}"}
|
|
elif method == "POST":
|
|
async with session.post(url, headers=headers, json=data) as resp:
|
|
if resp.status == 200:
|
|
return await resp.json()
|
|
return {"error": f"Status {resp.status}"}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
# Routes
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if request.method == 'POST':
|
|
password = request.form.get('password')
|
|
if password == WEB_PASSWORD:
|
|
session['logged_in'] = True
|
|
return redirect(url_for('index'))
|
|
else:
|
|
return render_template('login.html', error="Invalid password")
|
|
return render_template('login.html')
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
session.pop('logged_in', None)
|
|
return redirect(url_for('login'))
|
|
|
|
@app.route('/')
|
|
@login_required
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
@app.route('/services')
|
|
@login_required
|
|
def services():
|
|
return render_template('services.html')
|
|
|
|
@app.route('/attacks')
|
|
@login_required
|
|
def attacks():
|
|
return render_template('attacks.html')
|
|
|
|
@app.route('/alerts')
|
|
@login_required
|
|
def alerts():
|
|
return render_template('alerts.html')
|
|
|
|
# API Endpoints
|
|
@app.route('/api/dashboard')
|
|
@login_required
|
|
def api_dashboard():
|
|
"""Get dashboard overview data"""
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
# Fetch data from all services
|
|
services_data = loop.run_until_complete(api_call(f"{CONTROLLER_API}/services"))
|
|
scoreboard_stats = loop.run_until_complete(api_call(f"{SCOREBOARD_API}/stats"))
|
|
telegram_stats = loop.run_until_complete(api_call(f"{TELEGRAM_API}/stats"))
|
|
|
|
return jsonify({
|
|
"services": services_data,
|
|
"scoreboard": scoreboard_stats,
|
|
"telegram": telegram_stats,
|
|
"timestamp": datetime.utcnow().isoformat()
|
|
})
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/services')
|
|
@login_required
|
|
def api_services():
|
|
"""Get services list"""
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
data = loop.run_until_complete(api_call(f"{CONTROLLER_API}/services"))
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/services/<int:service_id>/action', methods=['POST'])
|
|
@login_required
|
|
def api_service_action(service_id):
|
|
"""Perform action on service"""
|
|
action = request.json.get('action')
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
data = loop.run_until_complete(
|
|
api_call(f"{CONTROLLER_API}/services/{service_id}/action", "POST", {"action": action})
|
|
)
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/services/<int:service_id>/update', methods=['PATCH'])
|
|
@login_required
|
|
def api_service_update(service_id):
|
|
"""Update service (e.g., alias)"""
|
|
updates = request.json
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
async def patch_call():
|
|
headers = {"Authorization": f"Bearer {SECRET_TOKEN}"}
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.patch(
|
|
f"{CONTROLLER_API}/services/{service_id}",
|
|
headers=headers,
|
|
json=updates
|
|
) as resp:
|
|
if resp.status == 200:
|
|
return await resp.json()
|
|
return {"error": f"Status {resp.status}"}
|
|
|
|
data = loop.run_until_complete(patch_call())
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/services/<int:service_id>/logs')
|
|
@login_required
|
|
def api_service_logs(service_id):
|
|
"""Get service logs"""
|
|
lines = request.args.get('lines', 100)
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
data = loop.run_until_complete(api_call(f"{CONTROLLER_API}/services/{service_id}/logs?lines={lines}"))
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/attacks')
|
|
@login_required
|
|
def api_attacks_list():
|
|
"""Get attacks list"""
|
|
limit = request.args.get('limit', 100)
|
|
our_attacks = request.args.get('our_attacks')
|
|
attacks_to_us = request.args.get('attacks_to_us')
|
|
|
|
url = f"{SCOREBOARD_API}/attacks?limit={limit}"
|
|
if our_attacks is not None:
|
|
url += f"&our_attacks={our_attacks}"
|
|
if attacks_to_us is not None:
|
|
url += f"&attacks_to_us={attacks_to_us}"
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
data = loop.run_until_complete(api_call(url))
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/attacks/by-service')
|
|
@login_required
|
|
def api_attacks_by_service():
|
|
"""Get attacks grouped by service"""
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
data = loop.run_until_complete(api_call(f"{SCOREBOARD_API}/attacks/by-service"))
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/alerts')
|
|
@login_required
|
|
def api_alerts_list():
|
|
"""Get alerts list"""
|
|
limit = request.args.get('limit', 50)
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
data = loop.run_until_complete(api_call(f"{SCOREBOARD_API}/alerts?limit={limit}"))
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
@app.route('/api/telegram/send', methods=['POST'])
|
|
@login_required
|
|
def api_telegram_send():
|
|
"""Send telegram message"""
|
|
message = request.json.get('message')
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
try:
|
|
data = loop.run_until_complete(
|
|
api_call(f"{TELEGRAM_API}/send", "POST", {"message": message})
|
|
)
|
|
return jsonify(data)
|
|
finally:
|
|
loop.close()
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=8000, debug=False)
|