Cobb Hayes
beead1d6b0
- requirements.txt: bump floors past known CVEs (flask>=2.3.2 fixes CVE-2023-30861, requests>=2.32.0 fixes CVE-2023-32681 + CVE-2024-35195, redis>=5.0 fixes CVE-2023-28858/9). - LICENSE: add MIT text (README claimed MIT but the file was missing). - /api/1/debug/redis-keys: require auth. Was unauthenticated info-disclosure on the LAN/AP side.
129 lines
5.1 KiB
Python
129 lines
5.1 KiB
Python
"""Flask app factory."""
|
|
import subprocess
|
|
from flask import Flask, request, jsonify
|
|
from . import config, db, forwarder
|
|
from .auth import get_device_serial, get_api_token, require_auth
|
|
from .routes import landmarks, gnss, status, frames, wigle
|
|
|
|
|
|
def create_app():
|
|
"""Create and configure the Flask app."""
|
|
app = Flask(__name__)
|
|
|
|
# Load config and init DB
|
|
config.load()
|
|
db.init()
|
|
|
|
# Register blueprints
|
|
app.register_blueprint(landmarks.bp)
|
|
app.register_blueprint(gnss.bp)
|
|
app.register_blueprint(status.bp)
|
|
app.register_blueprint(frames.bp)
|
|
app.register_blueprint(wigle.bp)
|
|
|
|
# ── PAIRING ENDPOINT ─────────────────────────────────────────────────────
|
|
@app.route('/pair')
|
|
@app.route('/api/1/pair')
|
|
def pair():
|
|
"""Pairing info for Varroa app."""
|
|
serial = get_device_serial()
|
|
return jsonify({
|
|
'serial': serial,
|
|
'version': '1.0',
|
|
'ap_ip': '10.77.0.1',
|
|
'api_port': 5000
|
|
})
|
|
|
|
# ── DEVICE INFO ──────────────────────────────────────────────────────────
|
|
@app.route('/api/1/info')
|
|
def device_info():
|
|
"""Device identity (unauthenticated)."""
|
|
serial = get_device_serial()
|
|
return jsonify({
|
|
'device_id': serial,
|
|
'version': '1.0',
|
|
'firmware': 'adacam',
|
|
'ap_ip': '10.77.0.1'
|
|
})
|
|
|
|
# ── WIFI STATUS/CONNECT ──────────────────────────────────────────────────
|
|
@app.route('/api/1/wifi/status')
|
|
def wifi_status():
|
|
"""Get WiFi client interface status (unauthenticated)."""
|
|
try:
|
|
result = subprocess.run(
|
|
['wpa_cli', '-i', 'wlp1s0f1', 'status'],
|
|
capture_output=True, text=True, timeout=5
|
|
)
|
|
lines = dict(
|
|
l.split('=', 1) for l in result.stdout.strip().split('\n') if '=' in l
|
|
)
|
|
return jsonify({
|
|
'ssid': lines.get('ssid', ''),
|
|
'ip': lines.get('ip_address', ''),
|
|
'state': lines.get('wpa_state', 'DISCONNECTED'),
|
|
'connected': lines.get('wpa_state') == 'COMPLETED'
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e), 'connected': False})
|
|
|
|
@app.route('/api/1/wifi/connect', methods=['POST'])
|
|
@require_auth
|
|
def wifi_connect():
|
|
"""Connect to a WiFi network (authenticated)."""
|
|
data = request.get_json()
|
|
ssid = data.get('ssid', '').strip()
|
|
password = data.get('password', '').strip()
|
|
if not ssid or not password:
|
|
return jsonify({'error': 'ssid and password required'}), 400
|
|
try:
|
|
result = subprocess.run(
|
|
['/usr/local/bin/adacam-wifi-connect', ssid, password],
|
|
capture_output=True, text=True, timeout=15
|
|
)
|
|
if result.returncode == 0:
|
|
return jsonify({'ok': True, 'message': f'Connecting to {ssid}'})
|
|
else:
|
|
return jsonify({'error': result.stderr or 'Failed'}), 500
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
# ── SSH CONTROL ──────────────────────────────────────────────────────────
|
|
@app.route('/api/1/ssh/status')
|
|
def ssh_status():
|
|
"""Get SSH daemon status (unauthenticated)."""
|
|
result = subprocess.run(
|
|
['systemctl', 'is-active', 'sshd'],
|
|
capture_output=True, text=True
|
|
)
|
|
return jsonify({'active': result.stdout.strip() == 'active'})
|
|
|
|
@app.route('/api/1/ssh/toggle', methods=['POST'])
|
|
@require_auth
|
|
def ssh_toggle():
|
|
"""Enable/disable SSH (authenticated)."""
|
|
data = request.get_json()
|
|
enable = data.get('enable', True)
|
|
subprocess.run(
|
|
['systemctl', 'start' if enable else 'stop', 'sshd'],
|
|
timeout=5
|
|
)
|
|
return jsonify({'ok': True, 'ssh_enabled': enable})
|
|
|
|
# ── DEBUG ENDPOINTS ────────────────────────────────────────────────────
|
|
@app.route('/api/1/debug/redis-keys')
|
|
@require_auth
|
|
def redis_keys():
|
|
"""Debug endpoint — list Redis keys for GPS/IMU troubleshooting."""
|
|
try:
|
|
import redis
|
|
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
|
|
keys = r.keys('*')
|
|
return jsonify({'keys': sorted(keys)})
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 503
|
|
|
|
# Start background forwarder thread
|
|
forwarder.start_retry_thread()
|
|
|
|
return app
|