⚠️ GeoIP database not found. Download GeoLite2-Country.mmdb and place it in the 'geoip/' folder to enable country lookup and automatic blocking.
📡 Traffic Analysis
Protocol Distribution
Top 5 Ports
🚨 Recent Threats
| Time |
Source IP |
Country |
Protocol |
Dest Port |
Threat Type |
Risk |
Action |
| Loading... |
| IP Address | Blocked Since | Reason | Action |
|---|
| Loading... |
'''
# ---------- Flask Routes ----------
@app.route('/')
def dashboard():
return render_template_string(HTML_TEMPLATE, geoip_available=geoip_available)
@app.route('/api/stats')
def api_stats():
total_events = db.get_total_event_count()
conn, cursor = db._get_connection()
cursor.execute('SELECT DISTINCT src_ip FROM events')
unique_src = len(cursor.fetchall())
blocked = db.get_blocked_ips()
return jsonify({
'total_events': total_events,
'unique_src': unique_src,
'blocked_count': len(blocked)
})
@app.route('/api/traffic_stats')
def api_traffic_stats():
events = db.get_recent_events(limit=2000)
protocol_counter = Counter()
port_counter = Counter()
for ev in events:
proto = ev[4]
port = ev[5]
protocol_counter[proto] += 1
if port:
port_counter[port] += 1
top_ports = dict(port_counter.most_common(5))
return jsonify({
'protocols': dict(protocol_counter),
'top_ports': top_ports
})
@app.route('/api/threats')
def api_threats():
events = db.get_recent_events(500)
threats = []
for e in events:
if e[7] or e[8] > 0:
src_ip = e[2]
country = None
if geoip and geoip.reader:
country = geoip.get_country_code(src_ip)
threats.append({
'time': e[1],
'src_ip': src_ip,
'country': country,
'protocol': e[4],
'port': e[5] if e[5] else None,
'threat_type': e[7] or 'MONITORED',
'risk': e[8],
'action': e[9]
})
return jsonify(threats[:100])
@app.route('/api/blocked_ips')
def api_blocked_ips():
blocked = db.get_blocked_ips()
return jsonify([{'ip': b[0], 'time': b[1], 'reason': b[2]} for b in blocked])
@app.route('/api/block', methods=['POST'])
def api_block():
data = request.get_json()
ip = data.get('ip', '').strip()
reason = data.get('reason', 'Manual block from dashboard')
if not ip:
return jsonify({'success': False, 'error': 'IP required'}), 400
success = block_ip_windows(ip, reason)
if success:
db.insert_blocked_ip(ip, reason)
return jsonify({'success': True})
else:
return jsonify({'success': False, 'error': 'Firewall rule failed'}), 500
@app.route('/api/unblock', methods=['POST'])
def api_unblock():
data = request.get_json()
ip = data.get('ip', '').strip()
if not ip:
return jsonify({'success': False, 'error': 'IP required'}), 400
success = unblock_ip_windows(ip)
if success:
conn, cursor = db._get_connection()
cursor.execute('DELETE FROM blocked_ips WHERE ip = ?', (ip,))
conn.commit()
return jsonify({'success': True})
else:
return jsonify({'success': False, 'error': 'Unblock failed'}), 500
@app.route('/api/download_blocked_ips_pdf')
def download_blocked_ips_pdf():
pdf_buffer = generate_blocked_ips_pdf()
return send_file(pdf_buffer, as_attachment=True, download_name='blocked_ips_log.pdf', mimetype='application/pdf')
@app.route('/api/download_all_ips_pdf')
def download_all_ips_pdf():
pdf_buffer = generate_all_ips_pdf()
return send_file(pdf_buffer, as_attachment=True, download_name='all_ip_traffic_log.pdf', mimetype='application/pdf')
@app.route('/api/download_pcap')
def download_pcap():
if not pcap_sniffer:
return jsonify({'error': 'PCAP capture not available'}), 500
packets = pcap_sniffer.get_pcap_buffer()
if not packets:
return jsonify({'error': 'No packets captured yet'}), 404
try:
with tempfile.NamedTemporaryFile(delete=False, suffix='.pcap') as tmp:
wrpcap(tmp.name, packets)
tmp_path = tmp.name
with open(tmp_path, 'rb') as f:
pcap_data = f.read()
os.unlink(tmp_path)
return send_file(io.BytesIO(pcap_data), as_attachment=True, download_name='capture.pcap', mimetype='application/vnd.tcpdump.pcap')
except Exception as e:
app.logger.error(f"PCAP export failed: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/geoip_status')
def geoip_status():
import main
return jsonify({'enabled': getattr(main, 'geoip_enabled', False)})
@app.route('/api/geoip_toggle', methods=['POST'])
def geoip_toggle():
import main
data = request.get_json()
main.geoip_enabled = data.get('enabled', False)
return jsonify({'success': True})
@app.route('/static/