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 = """
Server Admin
Server Admin
{% for label, _ in commands %}
{% endfor %}
{% if output is not none %}
Output:
{{ output }}
{% endif %}
"""
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)