1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
import functools
import os
import paramiko
from flask import Flask, Response, render_template_string, request
app = Flask(__name__)
ADMIN_USER = os.environ["ADMIN_USER"]
ADMIN_PASS = os.environ["ADMIN_PASS"]
SSH_HOST = os.environ.get("SSH_HOST", "host.docker.internal")
SSH_PORT = int(os.environ.get("SSH_PORT", "22"))
SSH_USER = os.environ.get("SSH_USER", "root")
SSH_KEY_PATH = os.environ.get("SSH_KEY_PATH", "/app/id_rsa")
COMMANDS = [
("Docker Status", "docker ps"),
("Restart Caddy", "docker compose -f /app/infra/caddy/docker-compose.yml restart"),
("Restart Mail", "docker compose -f /app/infra/mail/docker-compose.yml restart"),
("Restart cgit", "docker compose -f /app/infra/cgit/docker-compose.yml restart"),
("Restart Admin", "docker compose -f /app/infra/hetzner-admin/docker-compose.yml restart"),
("Restart Claude Sessions", "bash /app/infra/claude/restart-sessions.sh"),
("Reboot Server", "reboot"),
]
COMMAND_MAP = {label: cmd for label, cmd in COMMANDS}
HTML = """<!DOCTYPE html>
<html>
<head><title>Server Admin</title></head>
<body>
<h3>Server Admin</h3>
{% for label, _ in commands %}
<form method="post" action="/run" style="display:inline;margin-right:4px">
<input type="hidden" name="cmd" value="{{ label }}">
{% if label == "Reboot Server" %}
<button onclick="return confirm('Really reboot?')">{{ label }}</button>
{% else %}
<button>{{ label }}</button>
{% endif %}
</form>
{% endfor %}
{% if output is not none %}
<h4>Output:</h4>
<pre>{{ output }}</pre>
{% endif %}
</body>
</html>"""
def requires_auth(f):
@functools.wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
if not auth or auth.username != ADMIN_USER or auth.password != ADMIN_PASS:
return Response(
"Unauthorized", 401, {"WWW-Authenticate": 'Basic realm="Admin"'}
)
return f(*args, **kwargs)
return decorated
def run_ssh(command):
key = paramiko.RSAKey.from_private_key_file(SSH_KEY_PATH)
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(SSH_HOST, port=SSH_PORT, username=SSH_USER, pkey=key, timeout=10)
try:
_, stdout, stderr = client.exec_command(command, timeout=30)
out = stdout.read().decode(errors="replace")
err = stderr.read().decode(errors="replace")
return (out + err).strip() or "(no output)"
finally:
client.close()
@app.route("/")
@requires_auth
def index():
return render_template_string(HTML, commands=COMMANDS, output=None)
@app.route("/run", methods=["POST"])
@requires_auth
def run():
label = request.form.get("cmd", "")
command = COMMAND_MAP.get(label)
if not command:
output = f"Unknown command: {label}"
else:
try:
output = run_ssh(command)
except Exception as e:
output = f"Error: {e}"
return render_template_string(HTML, commands=COMMANDS, output=output)
|
