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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
import functools
import json
import os
import urllib.request
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")
STATUS_API = os.environ.get("STATUS_API", "http://status-app:8080/api/status")
# /status ページでコンテナ一覧を取るコマンド（run-command.sh のホワイトリストに要登録）
STATUS_COMMAND = "docker ps -a --format '{{.Names}}|{{.Status}}'"
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>
<p><a href="/status">Status</a></p>
{% for label, _ in commands %}
<form method="post" action="/run" style="display:block;margin-bottom:8px">
<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>"""
STATUS_HTML = """<!DOCTYPE html>
<html>
<head><title>Server Status</title><meta http-equiv="refresh" content="60"></head>
<body>
<h3>Server Status</h3>
<p><a href="/">← Admin</a></p>
<h4>Services</h4>
{% if services is none %}
<pre>{{ services_error }}</pre>
{% else %}
<table border="1" cellpadding="4" style="border-collapse:collapse">
<tr><th>Service</th><th>State</th><th>Detail</th><th>Latency</th><th>Uptime</th></tr>
{% for s in services %}
<tr><td>{{ s.name }}</td>
<td style="color:{{ 'green' if s.up else 'red' }}">{{ 'UP' if s.up else 'DOWN' }}</td>
<td>{{ s.detail }}</td><td>{{ s.ms }} ms</td><td>{{ s.uptime }}%</td></tr>
{% endfor %}
</table>
{% endif %}
<h4>Containers</h4>
{% if containers is none %}
<pre>{{ containers_error }}</pre>
{% else %}
<table border="1" cellpadding="4" style="border-collapse:collapse">
<tr><th>Container</th><th>Status</th></tr>
{% for name, st in containers %}
<tr><td>{{ name }}</td>
<td style="color:{{ 'green' if st.startswith('Up') and '(unhealthy)' not in st else 'red' }}">{{ st }}</td></tr>
{% endfor %}
</table>
{% 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)
@app.route("/status")
@requires_auth
def status():
services = services_error = containers = containers_error = None
try:
with urllib.request.urlopen(STATUS_API, timeout=5) as r:
services = json.load(r)["services"]
except Exception as e:
services_error = f"status-app に接続できない: {e}"
try:
out = run_ssh(STATUS_COMMAND)
containers = sorted(
tuple(line.split("|", 1)) for line in out.splitlines() if "|" in line
)
except Exception as e:
containers_error = f"Error: {e}"
return render_template_string(
STATUS_HTML,
services=services,
services_error=services_error,
containers=containers,
containers_error=containers_error,
)
|
