ad-infr-control/web/app.py

"""
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>/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)