Initial: Mesh-Skills, Configs, Scripts
This commit is contained in:
root
@@ -0,0 +1,89 @@
|
||||
---
|
||||
name: jit-ssh-access-portal
|
||||
description: "Web login + Telegram 2FA + time-boxed UFW SSH opening."
|
||||
version: 1.0.0
|
||||
platforms: [linux]
|
||||
---
|
||||
|
||||
# JIT SSH Access Portal (Telegram 2FA + Time-boxed Firewall)
|
||||
|
||||
Opens a web endpoint (`ssh.<domain>.de`) with password login, then a 6-digit **Telegram 2FA**
|
||||
code. On success, the backend opens **Port 22 in UFW only for the requesting IP** for a bounded
|
||||
time (default 1h) and auto-closes it. SSH stays closed-for-the-world otherwise.
|
||||
|
||||
## Why (security model)
|
||||
|
||||
- SSH is **closed by default** — no "22 Anywhere" rule.
|
||||
- 2FA over Telegram is **out-of-band** (a channel separate from the web login).
|
||||
- Firewall opening is **time-boxed** (`expiry` + background auto-close) → no permanent hole.
|
||||
- Passwords stored **hashed** (werkzeug `generate_password_hash`); first-use password forces change.
|
||||
|
||||
## Components
|
||||
|
||||
```
|
||||
nginx vhost ssh.<domain>.de (SSL) → reverse proxy → Flask app on 127.0.0.1:8091
|
||||
Flask app:
|
||||
- SQLite users table (username, pw_hash, pw_changed, failed, lockout_until)
|
||||
- POST / login (Erstpasswort → redirect zum PW-Wechsel; danach → /2fa)
|
||||
- POST /change-pw one-time password change (min length, confirm)
|
||||
- POST /2fa verify 6-digit code, `ufw allow from <ip> to any port 22` (1h)
|
||||
- background thread `_close_expired()` → `ufw delete allow ...` after expiry
|
||||
- `/srv` session carries `2fa_code`, `2fa_ts`, `2fa_attempts`
|
||||
```
|
||||
|
||||
Full reference implementation: `references/app.py` (drop-in `/opt/sshportal/app.py`).
|
||||
Deploy as a **systemd service** (survives reboot, overrides the VM-side "can't background" limit).
|
||||
|
||||
## Security anchors (non-negotiable)
|
||||
|
||||
1. **First password is one-time**: set `pw_changed=0` on first user; force change before 2FA.
|
||||
2. **2FA TTL + attempt cap**: code valid ~120s, max 3 attempts, then lock (log + Telegram alert).
|
||||
3. **Client IP source**: read `X-Real-IP`/`X-Forwarded-For` (set by nginx `proxy_set_header`),
|
||||
NOT `remote_addr` directly (would be the proxy).
|
||||
4. **Brute-force**: track `failed`; lockout window after N failures. **ACHTUNG (2026-08-04):** `users`-Tabelle hat `failed`/`lockout_until`-Spalten, und die `login()`-Route muss sie WIRKLICH nutzen — eine frühere Version hatte sie angelegt, aber nie abgefragt (Login prüfte nur Passwort). Zusätzlich IP-basiertes In-Memory-Lockout (`_IP_FAIL`), da Angriffe mit leerem/fremden Benutzernamen am User-Lockout vorbeigehen. Nach Nutzung von `login()`: echten User auf `failed=0, lockout_until=0` zurücksetzen, da Tests den Zähler hochziehen.
|
||||
5. **Rate-limit / TLS only**: only serve behind https (certbot `--redirect`).
|
||||
6. **Logging**: every login / pw-change / 2FA / open / close → local log + mesh event log.
|
||||
|
||||
## Pitfalls
|
||||
|
||||
- `render_template_string` escapes HTML by default → wrap dynamic HTML content in `Markup(...)`
|
||||
or pass plain text messages separately, or the form renders as escaped text (no `<h1>`).
|
||||
- Launching the Flask app with `nohup ... &` via the agent shell is blocked ("can't background
|
||||
inside gateway") → use a **systemd unit**, not a shell background process.
|
||||
- `ufw` interactive confirm: when deleting by number `ufw delete N` prompts → pipe `yes |` or
|
||||
delete by full rule string.
|
||||
- UFW rule dedupe: `ufw_allow` should skip if `ufw status` already shows `from <ip>`.
|
||||
- After the requesting IP is external (e.g. the browser came over the VPS public IP), the opened
|
||||
rule targets that external IP — if the real SSH client is on ZeroTier, the browser IP won't
|
||||
match. Decide whether the portal serves ZT-internal clients (requesting IP = ZT IP) or external.
|
||||
- **End-to-end test mutates the DB.** Running the full login→pw-change flow during testing
|
||||
overwrites the stored hash and sets `pw_changed=1`. If you don't reset afterward, the user's
|
||||
real first login FAILS with "falsches Passwort". After any E2E test, reset to first-use state:
|
||||
```bash
|
||||
sqlite3 /opt/sshportal/portal.db \
|
||||
"UPDATE users SET pw_hash='<bcrypt of first-pw>', pw_changed=0 WHERE username='mehlzoerwer';"
|
||||
```
|
||||
(or re-run the app's `ensure_first_user` logic). Essentially: a "password changed" side effect
|
||||
from testing must be rolled back before declaring the portal ready.
|
||||
|
||||
## Firewall pre-condition
|
||||
|
||||
Before enabling a JIT portal, tighten SSH so it is NOT "Anywhere". Restrict to the trusted
|
||||
ZeroTier mesh IPs, and close leftover WG/NFS ports:
|
||||
```bash
|
||||
ufw delete allow 22/tcp
|
||||
for ip in <trusted-ips>; do ufw allow from $ip to any port 22 proto tcp; done
|
||||
systemctl stop rpcbind rpcbind.socket && systemctl disable rpcbind rpcbind.socket
|
||||
```
|
||||
Public reachable surface should be **only 80 + 443**.
|
||||
|
||||
## Verification
|
||||
|
||||
- `curl https://ssh.<domain>.de/` → 200 login page.
|
||||
- First login (first PW) → `302 /change-pw`; set new PW → "Passwort geändert" log.
|
||||
- Second login → `302 /2fa`; backend logs "2FA-Code gesendet" and Telegram delivers a 6-digit code.
|
||||
- Submit correct code → `ufw status` shows `22/tcp ALLOW <ip> # SSH JIT Portal`; wrong code ×3 → lock.
|
||||
- After expiry → rule removed (test `ufw_allow`/`ufw_deny` on a throwaway IP end-to-end).
|
||||
|
||||
Full safe E2E test recipe for the login/lockout logic (throwaway user, IP-reset-before-DB-branch,
|
||||
cleanup): `references/safe-e2e-login-testing.md`.
|
||||
@@ -0,0 +1,201 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Drop-in reference: /opt/sshportal/app.py
|
||||
SSH-Portal mit Telegram-2FA und JIT-Firewall-Freischaltung.
|
||||
Login mehlzoerwer -> Erstpasswort -> PW-Wechsel-Pflicht -> 2. Login: PW+Telegram-Code
|
||||
-> UFW oeffnet Port 22 fuer angefragte IP (1 Stunde) -> Auto-Close.
|
||||
Abhaengigkeiten (venv): flask, werkzeug, requests.
|
||||
Start via systemd unit (NICHT nohup & im Agent-Shell).
|
||||
"""
|
||||
import os, re, time, secrets, sqlite3, subprocess, threading
|
||||
from pathlib import Path
|
||||
from flask import Flask, request, render_template_string, session, redirect, url_for
|
||||
from werkzeug.security import generate_password_hash, check_password_hash
|
||||
import requests
|
||||
|
||||
PORTAL_DIR = Path("/opt/sshportal")
|
||||
DB = PORTAL_DIR / "portal.db"
|
||||
LOG = PORTAL_DIR / "portal.log"
|
||||
ALLOW_TIME = 3600 # 1 Stunde offen
|
||||
MAX_2FA_ATTEMPTS = 3
|
||||
CODE_TTL = 120 # 2 Minuten gueltig
|
||||
|
||||
ENV_FILE = "/root/.hermes/.env"
|
||||
def _env(key):
|
||||
try:
|
||||
txt = Path(ENV_FILE).read_text()
|
||||
m = re.search(key + r"\s*=\s*(\S+)", txt)
|
||||
return m.group(1).strip() if m else ""
|
||||
except Exception:
|
||||
return ""
|
||||
TG_TOKEN = _env("TELEGRAM_BOT_TOKEN")
|
||||
TG_CHAT = _env("TELEGRAM_HOME_CHANNEL") or "6417835766"
|
||||
|
||||
app = Flask(__name__)
|
||||
app.secret_key = secrets.token_hex(32)
|
||||
|
||||
def log(msg):
|
||||
with open(LOG, "a") as f:
|
||||
f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}\n")
|
||||
|
||||
def db():
|
||||
c = sqlite3.connect(DB); c.row_factory = sqlite3.Row; return c
|
||||
|
||||
def init_db():
|
||||
c = db()
|
||||
c.execute("""CREATE TABLE IF NOT EXISTS users(
|
||||
id INTEGER PRIMARY KEY, username TEXT UNIQUE, pw_hash TEXT,
|
||||
pw_changed INTEGER DEFAULT 0, failed INTEGER DEFAULT 0, lockout_until INTEGER DEFAULT 0)""")
|
||||
c.execute("""CREATE TABLE IF NOT EXISTS sessions(
|
||||
ip TEXT, status TEXT, ts INTEGER, expiry INTEGER)""")
|
||||
c.commit(); c.close()
|
||||
|
||||
def ensure_first_user():
|
||||
c = db()
|
||||
row = c.execute("SELECT * FROM users WHERE username='mehlzoerwer'").fetchone()
|
||||
if not row:
|
||||
c.execute("INSERT INTO users(username,pw_hash,pw_changed) VALUES(?,?,0)",
|
||||
("mehlzoerwer", generate_password_hash("St230918kM@vps")))
|
||||
c.commit(); log("Erstbenutzer angelegt (Erstpasswort aktiv)")
|
||||
c.close()
|
||||
|
||||
def send_telegram(text):
|
||||
try:
|
||||
requests.post(f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage",
|
||||
json={"chat_id": TG_CHAT, "text": text}, timeout=8)
|
||||
except Exception as e:
|
||||
log(f"Telegram-Fehler: {e}")
|
||||
|
||||
def ufw_allow(ip):
|
||||
r = subprocess.run(["ufw","status"], capture_output=True, text=True)
|
||||
if f"from {ip}" in r.stdout:
|
||||
return False
|
||||
subprocess.run(["ufw","allow","from",ip,"to","any","port","22","proto","tcp",
|
||||
"comment","SSH JIT Portal"], check=False)
|
||||
return True
|
||||
|
||||
def ufw_deny(ip):
|
||||
subprocess.run(["ufw","delete","allow","from",ip,"to","any","port","22","proto","tcp",
|
||||
"comment","SSH JIT Portal"], check=False)
|
||||
|
||||
def client_ip():
|
||||
return request.headers.get("X-Real-IP") or request.remote_addr
|
||||
|
||||
def _close_expired():
|
||||
while True:
|
||||
try:
|
||||
c = db(); now = int(time.time())
|
||||
for r in c.execute("SELECT ip FROM sessions WHERE status='open' AND expiry<=?", (now,)):
|
||||
ufw_deny(r["ip"])
|
||||
c.execute("UPDATE sessions SET status='closed' WHERE ip=? AND status='open'", (r["ip"],))
|
||||
log(f"Auto-Close: SSH geschlossen fuer {r['ip']}")
|
||||
c.commit(); c.close()
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(60)
|
||||
|
||||
threading.Thread(target=_close_expired, daemon=True).start()
|
||||
|
||||
HTML = """<!doctype html><html lang="de"><head><meta charset="utf-8"><title>SSH-Zugangsportal</title>
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0"></head>
|
||||
<body style="font-family:system-ui;max-width:400px;margin:80px auto;padding:0 20px">
|
||||
{{ message }}{{ content }}</body></html>"""
|
||||
|
||||
def render(content, msg=None, msgtype="info"):
|
||||
from markupsafe import Markup
|
||||
m = ""
|
||||
if msg:
|
||||
m = f'<div style="background:{ "#e6f9ef;color:#0a7a4a" if msgtype=="ok" else "#fdecea;color:#b3261e" };padding:10px;border-radius:5px;margin:10px 0">{msg}</div>'
|
||||
return render_template_string(HTML, message=Markup(m), content=Markup(content))
|
||||
|
||||
FORM_LOGIN = '<h1>SSH-Zugang</h1><form method="post">' + \
|
||||
'<p><input name="username" placeholder="Benutzer" required style="width:100%;padding:10px"></p>' + \
|
||||
'<p><input name="password" type="password" placeholder="Passwort" required style="width:100%;padding:10px"></p>' + \
|
||||
'<button style="width:100%;padding:11px;background:#10b981;color:#fff;border:0;border-radius:5px">Anmelden</button></form>'
|
||||
|
||||
FORM_PW = '<h1>Neues Passwort</h1><form method="post">' + \
|
||||
'<p><input name="new1" type="password" placeholder="Neues Passwort" required style="width:100%;padding:10px"></p>' + \
|
||||
'<p><input name="new2" type="password" placeholder="Wiederholen" required style="width:100%;padding:10px"></p>' + \
|
||||
'<button style="width:100%;padding:11px;background:#10b981;color:#fff;border:0;border-radius:5px">Speichern</button></form>'
|
||||
|
||||
FORM_2FA = '<h1>2-Faktor</h1><p>Ein 6-stelliger Code wurde dir per Telegram zugesendet. Gültig 2 Minuten.</p>' + \
|
||||
'<form method="post"><input name="code" placeholder="6-stelliger Code" required style="width:100%;padding:10px">' + \
|
||||
'<button style="width:100%;padding:11px;background:#10b981;color:#fff;border:0;border-radius:5px;margin-top:8px">Bestätigen</button></form>'
|
||||
|
||||
@app.route("/", methods=["GET","POST"])
|
||||
def login():
|
||||
if request.method == "POST":
|
||||
username = request.form.get("username","").strip()
|
||||
password = request.form.get("password","")
|
||||
ip = client_ip()
|
||||
c = db()
|
||||
user = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||||
if not user or not check_password_hash(user["pw_hash"], password):
|
||||
log(f"FEHL_LOGIN {username} von {ip}")
|
||||
return render(FORM_LOGIN, "Falscher Benutzer oder Passwort.", "err")
|
||||
session["user"] = username
|
||||
if not user["pw_changed"]:
|
||||
log(f"PW-WECHSEL noetig {username} von {ip}")
|
||||
return redirect(url_for("change_pw"))
|
||||
session["pending_2fa_ip"] = ip
|
||||
code = f"{secrets.randbelow(1000000):06d}"
|
||||
session["2fa_code"] = code; session["2fa_ts"] = int(time.time()); session["2fa_attempts"] = 0
|
||||
send_telegram(f"SSH-Portal 2FA fuer Login von {ip}\nCode: {code}\nGueltig 2 Min.")
|
||||
log(f"2FA-Code an Chat {TG_CHAT} fuer Login von {ip}")
|
||||
return redirect(url_for("twofa"))
|
||||
return render(FORM_LOGIN)
|
||||
|
||||
@app.route("/change-pw", methods=["GET","POST"])
|
||||
def change_pw():
|
||||
if "user" not in session:
|
||||
return redirect(url_for("login"))
|
||||
if request.method == "POST":
|
||||
new1 = request.form.get("new1",""); new2 = request.form.get("new2","")
|
||||
if new1 != new2 or len(new1) < 10:
|
||||
return render(FORM_PW, "Passwoerter mismatch oder zu kurz (min 10).", "err")
|
||||
c = db()
|
||||
c.execute("UPDATE users SET pw_hash=?, pw_changed=1 WHERE username=?",
|
||||
(generate_password_hash(new1), session["user"]))
|
||||
c.commit(); c.close()
|
||||
log(f"Passwort geaendert fuer {session['user']}")
|
||||
send_telegram(f"Passwort fuer {session['user']} geaendert.")
|
||||
return render("<h1>Passwort geaendert</h1><p>Bitte neu einloggen. <a href='/'>Login</a></p>",
|
||||
"Passwort erfolgreich geaendert.", "ok")
|
||||
return render(FORM_PW)
|
||||
|
||||
@app.route("/2fa", methods=["GET","POST"])
|
||||
def twofa():
|
||||
if "2fa_code" not in session:
|
||||
return redirect(url_for("login"))
|
||||
if request.method == "POST":
|
||||
entered = request.form.get("code","").strip()
|
||||
ip = session["pending_2fa_ip"]
|
||||
if int(time.time()) - session["2fa_ts"] > CODE_TTL:
|
||||
session.pop("2fa_code", None)
|
||||
return render("", "Code abgelaufen. Bitte neu anmelden.", "err")
|
||||
if entered == session["2fa_code"]:
|
||||
added = ufw_allow(ip)
|
||||
c = db()
|
||||
c.execute("INSERT INTO sessions(ip,status,ts,expiry) VALUES(?,?,?,?)",
|
||||
(ip,"open",int(time.time()),int(time.time())+ALLOW_TIME))
|
||||
c.commit(); c.close()
|
||||
session.pop("2fa_code", None)
|
||||
log(f"SSH-Freischaltung fuer {ip} (1h)")
|
||||
send_telegram(f"SSH-Freischaltung fuer {ip} aktiv (1 Stunde).")
|
||||
return render(f"<h1>Zugriff freigeschaltet</h1><p>SSH auf Port 22 fuer <b>{ip}</b> fuer 1 Stunde.</p>",
|
||||
"Freigeschaltet!", "ok")
|
||||
else:
|
||||
session["2fa_attempts"] = session.get("2fa_attempts",0)+1
|
||||
log(f"FALSCHER 2FA-Code von {ip} (Versuch {session['2fa_attempts']})")
|
||||
if session["2fa_attempts"] >= MAX_2FA_ATTEMPTS:
|
||||
session.pop("2fa_code", None)
|
||||
send_telegram(f"3x falscher 2FA-Code von {ip} — gesperrt.")
|
||||
return render("", "Zu viele Fehlversuche. Login gesperrt.", "err")
|
||||
return render(FORM_2FA, f"Falscher Code. Versuche uebrig: {MAX_2FA_ATTEMPTS-session['2fa_attempts']}", "err")
|
||||
return render(FORM_2FA)
|
||||
|
||||
if __name__ == "__main__":
|
||||
PORTAL_DIR.mkdir(exist_ok=True)
|
||||
init_db(); ensure_first_user()
|
||||
log("SSH-Portal gestartet")
|
||||
app.run(host="127.0.0.1", port=8091)
|
||||
@@ -0,0 +1,53 @@
|
||||
# Sicheres E2E-Testen der Login-/Lockout-Logik (ohne echten User zu zerstören)
|
||||
|
||||
Der Bonuspitfall: Tests der Login-/Brute-Force-Logik dürfen den **echten
|
||||
Benutzeraccount** (`mehlzoerwer`) nicht verändern, sonst ist der echte Login
|
||||
danach defekt (Skill-Pitfall "E2E-Test mutiert die DB").
|
||||
|
||||
## Muster (validiert 2026-08-04)
|
||||
|
||||
1. **IP-Lockout-Zweig testen** → mit einem **fremden/leeren Benutzernamen**
|
||||
laufen lassen. Fremder User trifft nur den In-Memory-`_IP_FAIL`-Zähler,
|
||||
KEIN DB-Update auf den echten Account:
|
||||
```bash
|
||||
for i in $(seq 1 6); do
|
||||
curl -sk -X POST https://ssh.<domain>.de/ \
|
||||
-d "username=foo&password=wrong$i"
|
||||
done
|
||||
tail portal.log # Versuch 6 → "IP-SPERRE aktiv für <ip>"
|
||||
```
|
||||
Prüfen, dass der echte User danach `failed=0, lockout_until=0` ist.
|
||||
|
||||
2. **DB-User-Lockout-Zweig testen** → **temporären Test-User** in die DB,
|
||||
nach dem Test wieder löschen:
|
||||
```bash
|
||||
./venv/bin/python -c "
|
||||
import sqlite3
|
||||
from werkzeug.security import generate_password_hash
|
||||
c=sqlite3.connect('portal.db')
|
||||
c.execute(\"INSERT OR REPLACE INTO users(username,pw_hash,pw_changed,failed,lockout_until)\
|
||||
VALUES('testuser',?,1,0,0)\", (generate_password_hash('TestPass123!'),))
|
||||
c.commit(); c.close()"
|
||||
```
|
||||
Dann 5x falsches PW für `testuser` → DB zeigt `failed=5, lockout_until=<ts>`,
|
||||
Log zeigt "USER-SPERRE aktiv".
|
||||
|
||||
3. **WICHTIG — In-Memory-IP-Sperre leert nur beim Service-Restart**:
|
||||
Nach Test 1 ist die Test-IP noch 15 min in `_IP_FAIL` gesperrt. Vor dem
|
||||
Testen des zweiten Zweigs `systemctl restart sshportal` (leert den
|
||||
In-Memory-Zähler), sonst kommen die neuen Requests gar nicht bis zur DB.
|
||||
|
||||
4. **Aufräumen** (immer, in einer Transaktion committen):
|
||||
```bash
|
||||
c.execute('DELETE FROM users WHERE username=?', ('testuser',))
|
||||
c.execute('UPDATE users SET failed=0, lockout_until=0 WHERE username=?', ('mehlzoerwer',))
|
||||
c.commit()
|
||||
```
|
||||
|
||||
## Fallstricke
|
||||
- `sqlite3`-CLI fehlt oft auf dem VPS → über das venv-Python lesen
|
||||
(`./venv/bin/python -c "..."`), nicht `sqlite3 portal.db`.
|
||||
- DB-Schema kann Alt-Spalten haben → `dict(row)` schlägt beim Print fehl
|
||||
("length 11; 2 is required"); explizite Spalten im SELECT wählen.
|
||||
- Desktop-`dict(rows)`-Print ist nur Reporting, nicht Teil des Deletes — ein
|
||||
Traceback dort heißt NICHT, dass der Delete fehlgeschlagen ist.
|
||||
Reference in New Issue
Block a user