72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
"""PiCopy – Blueprint: /api/wireguard/*."""
|
||
|
||
import re
|
||
import threading
|
||
|
||
from flask import Blueprint, jsonify, request
|
||
|
||
from picopy.config import load_cfg, save_cfg
|
||
from picopy.wireguard import (
|
||
wg_state, wg_lock, WG_CONF,
|
||
wg_connect, wg_disconnect,
|
||
wg_install, wg_uninstall,
|
||
wg_save_config,
|
||
)
|
||
|
||
wireguard_bp = Blueprint('wireguard', __name__)
|
||
|
||
|
||
@wireguard_bp.route('/api/wireguard/config', methods=['GET', 'POST'])
|
||
def r_wg_config():
|
||
if request.method == 'POST':
|
||
data = request.get_json(force=True)
|
||
content = data.get('content', '')
|
||
if not content.strip():
|
||
return jsonify(error='Konfiguration ist leer'), 400
|
||
ok, err = wg_save_config(content)
|
||
if not ok:
|
||
return jsonify(error=err), 500
|
||
auto = data.get('auto')
|
||
if auto is not None:
|
||
c = load_cfg()
|
||
c['wireguard_auto'] = bool(auto)
|
||
save_cfg(c)
|
||
with wg_lock:
|
||
wg_state['has_config'] = True
|
||
return jsonify(ok=True)
|
||
if WG_CONF.exists():
|
||
content = WG_CONF.read_text(encoding='utf-8')
|
||
masked = re.sub(r'(PrivateKey\s*=\s*)(.+)', r'\1****', content)
|
||
return jsonify(exists=True, config=masked)
|
||
return jsonify(exists=False, config='')
|
||
|
||
|
||
@wireguard_bp.route('/api/wireguard/connect', methods=['POST'])
|
||
def r_wg_connect():
|
||
threading.Thread(target=wg_connect, daemon=True).start()
|
||
return jsonify(ok=True, msg='Verbindungsversuch gestartet')
|
||
|
||
|
||
@wireguard_bp.route('/api/wireguard/disconnect', methods=['POST'])
|
||
def r_wg_disconnect():
|
||
ok = wg_disconnect()
|
||
return jsonify(ok=ok)
|
||
|
||
|
||
@wireguard_bp.route('/api/wireguard/install', methods=['POST'])
|
||
def r_wg_install():
|
||
with wg_lock:
|
||
if wg_state['pkg_running']:
|
||
return jsonify(error='Bereits aktiv'), 400
|
||
threading.Thread(target=wg_install, daemon=True).start()
|
||
return jsonify(ok=True)
|
||
|
||
|
||
@wireguard_bp.route('/api/wireguard/uninstall', methods=['POST'])
|
||
def r_wg_uninstall():
|
||
with wg_lock:
|
||
if wg_state['pkg_running']:
|
||
return jsonify(error='Bereits aktiv'), 400
|
||
threading.Thread(target=wg_uninstall, daemon=True).start()
|
||
return jsonify(ok=True)
|