企业级堡垒机实战：从架构设计到 SSH 会话审计的完整代码实现

运维工程师 -> 直连生产服务器（SSH Key）
                    |
                    └── 谁在什么时候做了什么？没人知道。

运维工程师 -> 堡垒机（认证/授权/审计）-> 目标服务器
                    |
                    └── 全部操作被记录、可回放、可追责

┌───────────────────────────────────────────┐
│ 1. 统一入口   所有运维流量经堡垒机          │
│ 2. 身份认证   密码 / SSH Key / MFA / SSO    │
│ 3. 授权控制   谁能访问哪些资产、哪些命令     │
│ 4. 操作审计   命令日志 + 会话录制            │
│ 5. 会话管控   实时监控 / 强制阻断 / 协同     │
│ 6. 合规报表   审计导出 / 责任追溯            │
└───────────────────────────────────────────┘

┌────────────────────────────────────────────────────────┐
│                        用户层                            │
│   Web 终端 / SSH Client / RDP Client / API              │
├────────────────────────────────────────────────────────┤
│                        接入层                            │
│   Nginx / LB / WebSocket / SSH Server / RDP Gateway     │
├────────────────────────────────────────────────────────┤
│                        认证层                            │
│   密码 / SSH Key / TOTP / LDAP / OIDC / SAML            │
├────────────────────────────────────────────────────────┤
│                        授权层                            │
│   RBAC / ABAC / 资产授权 / 命令策略 / 时间窗口           │
├────────────────────────────────────────────────────────┤
│                        代理层                            │
│   SSH Proxy / RDP Proxy / DB Proxy / K8s Proxy          │
├────────────────────────────────────────────────────────┤
│                        审计层                            │
│   命令日志 / 会话录制 / 文件传输审计 / 回放              │
├────────────────────────────────────────────────────────┤
│                        数据层                            │
│   PostgreSQL / Redis / 对象存储 / 密钥管理（KMS/Vault）  │
├────────────────────────────────────────────────────────┤
│                        运维层                            │
│   日志 / 监控 / 告警 / 备份 / HA                        │
└────────────────────────────────────────────────────────┘

Client  <--SSH-->  Bastion  <--SSH-->  Target

import paramiko

# 1. 作为 SSH Server 接受客户端
# 2. 作为 SSH Client 连接目标
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname, username, password)

# 3. 打开 channel，双向转发
chan = client.invoke_shell()

enterprise-bastion/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── crypto.py
│   ├── auth.py
│   ├── rbac.py
│   ├── assets.py
│   ├── ssh_proxy.py
│   ├── audit.py
│   ├── policy.py
│   ├── recorder.py
│   └── playback.py
├── tests/
│   └── test_bastion.py
├── requirements.txt
└── Dockerfile

pip install fastapi uvicorn sqlmodel paramiko cryptography pyotp pytest httpx

fastapi
uvicorn[standard]
sqlmodel
paramiko
cryptography
pyotp
pytest
httpx

from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True)
    password_hash: str
    totp_secret: Optional[str] = None
    role: str = "operator"          # admin / operator / auditor / viewer
    enabled: bool = True
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Asset(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    hostname: str
    port: int = 22
    protocol: str = "ssh"           # ssh / rdp / vnc / db
    group: str = "default"
    enabled: bool = True
    created_at: datetime = Field(default_factory=datetime.utcnow)


class AssetAccount(SQLModel, table=True):
    """
    目标服务器上的账号，凭证加密存储。
    """
    id: Optional[int] = Field(default=None, primary_key=True)
    asset_id: int = Field(index=True)
    username: str
    auth_type: str = "password"     # password / key
    credential_enc: str             # 加密后的密码或私钥
    created_at: datetime = Field(default_factory=datetime.utcnow)


class Grant(SQLModel, table=True):
    """
    授权关系：某用户对某资产在某时间段有权限。
    """
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    asset_id: int = Field(index=True)
    account_id: int
    can_shell: bool = True
    can_sftp: bool = False
    valid_from: datetime = Field(default_factory=datetime.utcnow)
    valid_to: Optional[datetime] = None


class CommandPolicy(SQLModel, table=True):
    """
    命令策略：黑名单或白名单。
    """
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    pattern: str                    # 正则
    action: str = "block"           # block / alert / log
    severity: str = "high"
    enabled: bool = True


class Session(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(index=True)
    asset_id: int = Field(index=True)
    account_id: int
    client_ip: str = ""
    started_at: datetime = Field(default_factory=datetime.utcnow)
    ended_at: Optional[datetime] = None
    status: str = "active"          # active / closed / blocked


class AuditCommand(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    session_id: int = Field(index=True)
    user_id: int
    asset_id: int
    command: str
    action: str = "allow"           # allow / block / alert
    severity: str = "info"
    created_at: datetime = Field(default_factory=datetime.utcnow)


class SessionRecording(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    session_id: int = Field(index=True, unique=True)
    path: str
    size_bytes: int = 0
    created_at: datetime = Field(default_factory=datetime.utcnow)

import os
import base64
import hashlib
import hmac
from cryptography.fernet import Fernet

# 生产环境请从 KMS / Vault 加载
MASTER_KEY = os.getenv("BASTION_MASTER_KEY", "")


def _get_fernet() -> Fernet:
    if not MASTER_KEY:
        raise RuntimeError("BASTION_MASTER_KEY 未配置")
    return Fernet(MASTER_KEY.encode())


def encrypt_credential(plaintext: str) -> str:
    return _get_fernet().encrypt(plaintext.encode()).decode()


def decrypt_credential(ciphertext: str) -> str:
    return _get_fernet().decrypt(ciphertext.encode()).decode()


def hash_password(password: str, salt: str | None = None) -> str:
    salt = salt or base64.b16encode(os.urandom(16)).decode()
    dk = hashlib.pbkdf2_hmac(
        "sha256",
        password.encode(),
        salt.encode(),
        200_000,
    )
    return f"pbkdf2_sha256$200000${salt}${base64.b64encode(dk).decode()}"


def verify_password(password: str, stored: str) -> bool:
    try:
        algo, iterations, salt, encoded = stored.split("$")
        dk = hashlib.pbkdf2_hmac(
            "sha256",
            password.encode(),
            salt.encode(),
            int(iterations),
        )
        return hmac.compare_digest(
            base64.b64encode(dk).decode(),
            encoded,
        )
    except Exception:
        return False


def generate_master_key() -> str:
    return Fernet.generate_key().decode()

python -c "from app.crypto import generate_master_key; print(generate_master_key())"
export BASTION_MASTER_KEY="生成的密钥"

from datetime import datetime, timedelta
from fastapi import HTTPException, Depends, Header
from jose import jwt, JWTError
from sqlmodel import Session, select
import pyotp

from app.db import get_session
from app.models import User
from app.crypto import verify_password

JWT_SECRET = "change-me-in-production"
JWT_ALG = "HS256"
JWT_EXPIRE_MINUTES = 60


def authenticate(
    session: Session,
    username: str,
    password: str,
    totp_code: str | None = None,
) -> User:
    user = session.exec(
        select(User).where(User.username == username)
    ).first()

    if not user or not user.enabled:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    if not verify_password(password, user.password_hash):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # MFA 校验
    if user.totp_secret:
        if not totp_code:
            raise HTTPException(status_code=401, detail="MFA code required")
        totp = pyotp.TOTP(user.totp_secret)
        if not totp.verify(totp_code, valid_window=1):
            raise HTTPException(status_code=401, detail="Invalid MFA code")

    return user


def issue_token(user: User) -> str:
    payload = {
        "sub": str(user.id),
        "username": user.username,
        "role": user.role,
        "exp": datetime.utcnow() + timedelta(minutes=JWT_EXPIRE_MINUTES),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALG)


def get_current_user(
    authorization: str = Header(...),
    session: Session = Depends(get_session),
) -> User:
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid auth header")

    token = authorization.replace("Bearer ", "")
    try:
        payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG])
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = session.get(User, int(payload["sub"]))
    if not user or not user.enabled:
        raise HTTPException(status_code=401, detail="User disabled")
    return user


def require_role(*roles):
    def checker(user: User = Depends(get_current_user)):
        if user.role not in roles:
            raise HTTPException(status_code=403, detail="Forbidden")
        return user
    return checker

from datetime import datetime
from fastapi import HTTPException
from sqlmodel import Session, select

from app.models import Grant, Asset, AssetAccount, User


def check_asset_access(
    session: Session,
    user: User,
    asset_id: int,
    need_sftp: bool = False,
) -> tuple[Asset, AssetAccount]:
    asset = session.get(Asset, asset_id)
    if not asset or not asset.enabled:
        raise HTTPException(status_code=404, detail="Asset not found")

    now = datetime.utcnow()
    grants = session.exec(
        select(Grant).where(
            Grant.user_id == user.id,
            Grant.asset_id == asset_id,
        )
    ).all()

    for g in grants:
        if g.valid_from > now:
            continue
        if g.valid_to and g.valid_to < now:
            continue
        if need_sftp and not g.can_sftp:
            continue
        if not need_sftp and not g.can_shell:
            continue

        account = session.get(AssetAccount, g.account_id)
        if not account:
            continue
        return asset, account

    raise HTTPException(status_code=403, detail="No permission for this asset")

import socket
import threading
import select
import paramiko

from app.crypto import decrypt_credential
from app.models import Asset, AssetAccount
from app.recorder import SessionRecorder
from app.policy import CommandAuditor


class SSHProxy:
    """
    最小 SSH 代理：
    - 作为 Client 连接目标
    - 转发 stdin/stdout
    - 调用 CommandAuditor 做命令审计
    - 调用 SessionRecorder 录制
    """

    def __init__(
        self,
        asset: Asset,
        account: AssetAccount,
        user_id: int,
        session_id: int,
        recorder: SessionRecorder,
        auditor: CommandAuditor,
    ):
        self.asset = asset
        self.account = account
        self.user_id = user_id
        self.session_id = session_id
        self.recorder = recorder
        self.auditor = auditor
        self.client = None
        self.channel = None
        self._stop = threading.Event()

    def connect(self):
        credential = decrypt_credential(self.account.credential_enc)

        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        if self.account.auth_type == "password":
            self.client.connect(
                hostname=self.asset.hostname,
                port=self.asset.port,
                username=self.account.username,
                password=credential,
                timeout=10,
                banner_timeout=10,
            )
        else:
            from io import StringIO
            key = paramiko.RSAKey.from_private_key(StringIO(credential))
            self.client.connect(
                hostname=self.asset.hostname,
                port=self.asset.port,
                username=self.account.username,
                pkey=key,
                timeout=10,
            )

        self.channel = self.client.invoke_shell(term="xterm")
        self.channel.settimeout(0.0)

    def bridge(self, client_sock: socket.socket):
        """
        双向转发：client <-> channel
        """
        try:
            while not self._stop.is_set():
                r, _, _ = select.select([client_sock, self.channel], [], [], 1.0)
                if client_sock in r:
                    data = client_sock.recv(4096)
                    if not data:
                        break
                    self.recorder.record_input(data)
                    self.auditor.feed(data)
                    self.channel.send(data)

                if self.channel in r:
                    out = self.channel.recv(4096)
                    if not out:
                        break
                    self.recorder.record_output(out)
                    client_sock.sendall(out)
        finally:
            self.close()

    def close(self):
        self._stop.set()
        try:
            if self.channel:
                self.channel.close()
            if self.client:
                self.client.close()
        except Exception:
            pass

import re
from datetime import datetime
from sqlmodel import Session, select

from app.models import CommandPolicy, AuditCommand


DEFAULT_BLACKLIST = [
    (r"\brm\s+-rf\s+/\b", "block", "critical"),
    (r"\bmkfs\b", "block", "critical"),
    (r"\bdd\s+if=.*of=/dev/", "block", "critical"),
    (r":\(\)\s*\{.*\};:", "block", "critical"),   # fork bomb
    (r"\bshutdown\b", "block", "high"),
    (r"\breboot\b", "block", "high"),
    (r"\bpasswd\s+root\b", "alert", "high"),
    (r"\biptables\s+-F\b", "alert", "high"),
    (r"\bchmod\s+777\s+/\b", "block", "high"),
]


class CommandAuditor:
    """
    逐字符累积输入流，按行解析命令并匹配策略。
    真实产品需要处理 tab 补全、箭头键、Ctrl+C 等控制字符。
    """

    def __init__(
        self,
        db_session: Session,
        session_id: int,
        user_id: int,
        asset_id: int,
    ):
        self.db = db_session
        self.session_id = session_id
        self.user_id = user_id
        self.asset_id = asset_id
        self.buffer = ""

        self.policies = self._load_policies()

    def _load_policies(self):
        rows = self.db.exec(
            select(CommandPolicy).where(CommandPolicy.enabled == True)  # noqa: E712
        ).all()

        policies = [(p.pattern, p.action, p.severity) for p in rows]
        if not policies:
            policies = DEFAULT_BLACKLIST
        return policies

    def feed(self, data: bytes):
        try:
            text = data.decode("utf-8", errors="ignore")
        except Exception:
            return

        for ch in text:
            if ch in ("\r", "\n"):
                self._evaluate(self.buffer.strip())
                self.buffer = ""
            elif ch == "\x7f":                # backspace
                self.buffer = self.buffer[:-1]
            elif ch == "\x03":                # Ctrl+C
                self.buffer = ""
            elif ch.isprintable():
                self.buffer += ch

    def _evaluate(self, command: str):
        if not command:
            return

        action = "allow"
        severity = "info"

        for pattern, act, sev in self.policies:
            if re.search(pattern, command):
                action = act
                severity = sev
                break

        log = AuditCommand(
            session_id=self.session_id,
            user_id=self.user_id,
            asset_id=self.asset_id,
            command=command,
            action=action,
            severity=severity,
        )
        self.db.add(log)
        self.db.commit()

        if action == "block":
            # 真实环境需要向 client 发送阻断提示并断开或拒绝执行
            raise PermissionError(f"Command blocked by policy: {command}")

import os
import json
import time
from pathlib import Path

RECORD_DIR = Path(os.getenv("BASTION_RECORD_DIR", "./recordings"))


class SessionRecorder:
    """
    录制格式：JSON Lines，每行一条事件。
    字段：
      t: 相对时间（秒）
      d: 方向（i=输入, o=输出）
      c: base64 或转义后的内容
    """

    def __init__(self, session_id: int):
        RECORD_DIR.mkdir(parents=True, exist_ok=True)
        self.session_id = session_id
        self.path = RECORD_DIR / f"session_{session_id}.jsonl"
        self.start = time.time()
        self.fp = open(self.path, "a", encoding="utf-8")

    def _write(self, direction: str, data: bytes):
        event = {
            "t": round(time.time() - self.start, 4),
            "d": direction,
            "c": data.decode("utf-8", errors="replace"),
        }
        self.fp.write(json.dumps(event, ensure_ascii=False) + "\n")
        self.fp.flush()

    def record_input(self, data: bytes):
        self._write("i", data)

    def record_output(self, data: bytes):
        self._write("o", data)

    def close(self):
        try:
            self.fp.close()
        except Exception:
            pass


def file_size(path: str) -> int:
    try:
        return os.path.getsize(path)
    except OSError:
        return 0

import json
import time
from pathlib import Path


def load_recording(path: str) -> list:
    events = []
    with open(path, "r", encoding="utf-8") as fp:
        for line in fp:
            line = line.strip()
            if not line:
                continue
            events.append(json.loads(line))
    return events


def render_text(path: str) -> str:
    """
    将录制还原成可读的文本流。
    """
    events = load_recording(path)
    out = []
    for e in events:
        if e["d"] == "i":
            out.append(f"\n[IN  {e['t']:>8.3f}s] {e['c']}")
        else:
            out.append(e["c"])
    return "".join(out)


def replay_timed(path: str, speed: float = 1.0):
    """
    按时间轴回放（用于终端演示）。
    """
    events = load_recording(path)
    prev_t = 0.0
    for e in events:
        delay = (e["t"] - prev_t) / speed
        if delay > 0:
            time.sleep(delay)
        prev_t = e["t"]
        prefix = "IN " if e["d"] == "i" else "OUT"
        print(f"[{prefix} {e['t']:.3f}] {e['c']}", end="")

import uuid
from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, WebSocket
from pydantic import BaseModel
from sqlmodel import Session, select

from app.db import init_db, get_session
from app.models import (
    User, Asset, AssetAccount, Grant, Session as SSHSession,
    AuditCommand, SessionRecording,
)
from app.auth import (
    authenticate, issue_token, get_current_user, require_role,
)
from app.crypto import hash_password, encrypt_credential
from app.rbac import check_asset_access
from app.recorder import SessionRecorder, file_size
from app.playback import render_text
import pyotp

app = FastAPI(title="Enterprise Bastion Host")


@app.on_event("startup")
def on_startup():
    init_db()


class LoginPayload(BaseModel):
    username: str
    password: str
    totp_code: str | None = None


class AssetPayload(BaseModel):
    name: str
    hostname: str
    port: int = 22
    protocol: str = "ssh"
    group: str = "default"


class AccountPayload(BaseModel):
    asset_id: int
    username: str
    auth_type: str = "password"
    credential: str


class GrantPayload(BaseModel):
    user_id: int
    asset_id: int
    account_id: int
    can_shell: bool = True
    can_sftp: bool = False
    valid_days: int = 30


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/auth/login")
def login(payload: LoginPayload, session: Session = Depends(get_session)):
    user = authenticate(session, payload.username, payload.password, payload.totp_code)
    token = issue_token(user)
    return {"access_token": token, "token_type": "bearer", "role": user.role}


@app.post("/users")
def create_user(
    username: str,
    password: str,
    role: str = "operator",
    enable_mfa: bool = False,
    session: Session = Depends(get_session),
    _: User = Depends(require_role("admin")),
):
    totp_secret = pyotp.random_base32() if enable_mfa else None
    user = User(
        username=username,
        password_hash=hash_password(password),
        role=role,
        totp_secret=totp_secret,
    )
    session.add(user)
    session.commit()
    session.refresh(user)

    result = {"user_id": user.id, "username": user.username, "role": user.role}
    if totp_secret:
        result["totp_secret"] = totp_secret
        result["otp_uri"] = pyotp.TOTP(totp_secret).provisioning_uri(
            name=username, issuer_name="EnterpriseBastion"
        )
    return result


@app.post("/assets")
def create_asset(
    payload: AssetPayload,
    session: Session = Depends(get_session),
    _: User = Depends(require_role("admin")),
):
    asset = Asset(**payload.dict())
    session.add(asset)
    session.commit()
    session.refresh(asset)
    return asset


@app.post("/accounts")
def create_account(
    payload: AccountPayload,
    session: Session = Depends(get_session),
    _: User = Depends(require_role("admin")),
):
    account = AssetAccount(
        asset_id=payload.asset_id,
        username=payload.username,
        auth_type=payload.auth_type,
        credential_enc=encrypt_credential(payload.credential),
    )
    session.add(account)
    session.commit()
    session.refresh(account)
    return {
        "id": account.id,
        "asset_id": account.asset_id,
        "username": account.username,
        "auth_type": account.auth_type,
    }


@app.post("/grants")
def create_grant(
    payload: GrantPayload,
    session: Session = Depends(get_session),
    _: User = Depends(require_role("admin")),
):
    grant = Grant(
        user_id=payload.user_id,
        asset_id=payload.asset_id,
        account_id=payload.account_id,
        can_shell=payload.can_shell,
        can_sftp=payload.can_sftp,
        valid_to=datetime.utcnow() + timedelta(days=payload.valid_days),
    )
    session.add(grant)
    session.commit()
    session.refresh(grant)
    return grant


@app.get("/assets/my")
def my_assets(
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    now = datetime.utcnow()
    grants = session.exec(
        select(Grant).where(Grant.user_id == user.id)
    ).all()

    result = []
    for g in grants:
        if g.valid_from > now:
            continue
        if g.valid_to and g.valid_to < now:
            continue
        asset = session.get(Asset, g.asset_id)
        if not asset or not asset.enabled:
            continue
        result.append({
            "asset_id": asset.id,
            "name": asset.name,
            "hostname": asset.hostname,
            "group": asset.group,
            "account_id": g.account_id,
            "can_shell": g.can_shell,
            "can_sftp": g.can_sftp,
            "valid_to": str(g.valid_to) if g.valid_to else None,
        })
    return result


@app.post("/sessions/start")
def start_session(
    asset_id: int,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    asset, account = check_asset_access(session, user, asset_id)

    ssh_session = SSHSession(
        user_id=user.id,
        asset_id=asset.id,
        account_id=account.id,
        client_ip="0.0.0.0",
    )
    session.add(ssh_session)
    session.commit()
    session.refresh(ssh_session)

    recorder = SessionRecorder(ssh_session.id)
    recorder.close()

    rec = SessionRecording(
        session_id=ssh_session.id,
        path=str(recorder.path),
    )
    session.add(rec)
    session.commit()

    return {
        "session_id": ssh_session.id,
        "asset": asset.name,
        "hostname": asset.hostname,
        "account": account.username,
        "websocket_url": f"/ws/terminal/{ssh_session.id}",
    }


@app.post("/sessions/{session_id}/close")
def close_session(
    session_id: int,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    s = session.get(SSHSession, session_id)
    if not s:
        raise HTTPException(status_code=404, detail="Session not found")
    if s.user_id != user.id and user.role not in ("admin", "auditor"):
        raise HTTPException(status_code=403, detail="Forbidden")

    s.status = "closed"
    s.ended_at = datetime.utcnow()
    session.add(s)
    session.commit()
    return {"session_id": s.id, "status": s.status}


@app.get("/audit/commands")
def audit_commands(
    session_id: int | None = None,
    user_id: int | None = None,
    action: str | None = None,
    user: User = Depends(require_role("admin", "auditor")),
    session: Session = Depends(get_session),
):
    stmt = select(AuditCommand)
    if session_id:
        stmt = stmt.where(AuditCommand.session_id == session_id)
    if user_id:
        stmt = stmt.where(AuditCommand.user_id == user_id)
    if action:
        stmt = stmt.where(AuditCommand.action == action)

    rows = session.exec(stmt.order_by(AuditCommand.id.desc()).limit(500)).all()
    return rows


@app.get("/audit/sessions")
def audit_sessions(
    user: User = Depends(require_role("admin", "auditor")),
    session: Session = Depends(get_session),
):
    rows = session.exec(
        select(SSHSession).order_by(SSHSession.id.desc()).limit(200)
    ).all()
    return rows


@app.get("/audit/sessions/{session_id}/recording")
def get_recording(
    session_id: int,
    user: User = Depends(require_role("admin", "auditor")),
    session: Session = Depends(get_session),
):
    rec = session.exec(
        select(SessionRecording).where(SessionRecording.session_id == session_id)
    ).first()
    if not rec:
        raise HTTPException(status_code=404, detail="Recording not found")
    return {"path": rec.path, "size": file_size(rec.path)}


@app.get("/audit/sessions/{session_id}/playback")
def get_playback(
    session_id: int,
    user: User = Depends(require_role("admin", "auditor")),
    session: Session = Depends(get_session),
):
    rec = session.exec(
        select(SessionRecording).where(SessionRecording.session_id == session_id)
    ).first()
    if not rec:
        raise HTTPException(status_code=404, detail="Recording not found")
    return {"text": render_text(rec.path)}

from fastapi import WebSocket, WebSocketDisconnect


@app.websocket("/ws/terminal/{session_id}")
async def terminal_ws(
    websocket: WebSocket,
    session_id: int,
):
    await websocket.accept()
    try:
        await websocket.send_text(
            "WebSocket terminal ready. "
            "生产环境需接入 SSHProxy 完成双向转发。\r\n"
        )
        while True:
            data = await websocket.receive_text()
            if data == "exit":
                break
            # TODO: 转发到 SSHProxy 的 channel
            await websocket.send_text(f"echo: {data}\r\n")
    except WebSocketDisconnect:
        pass
    finally:
        await websocket.close()

import os
import pytest
from fastapi.testclient import TestClient

os.environ["BASTION_MASTER_KEY"] = "Z0m8vQ3n7pL1xR5tY9uI2oP4aS6dF8gH0jK2lM4nB6c="
os.environ["DATABASE_URL"] = "sqlite:///./test_bastion.db"

from app.main import app  # noqa: E402

client = TestClient(app)


def _bootstrap_admin():
    r = client.post(
        "/users",
        params={
            "username": "admin",
            "password": "admin123",
            "role": "admin",
        },
    )
    assert r.status_code == 200

    r = client.post(
        "/auth/login",
        json={"username": "admin", "password": "admin123"},
    )
    assert r.status_code == 200
    return r.json()["access_token"]


def test_health():
    assert client.get("/health").json()["status"] == "ok"


def test_full_flow():
    token = _bootstrap_admin()
    headers = {"Authorization": f"Bearer {token}"}

    # 创建资产
    r = client.post(
        "/assets",
        json={
            "name": "web-01",
            "hostname": "10.0.0.10",
            "port": 22,
        },
        headers=headers,
    )
    asset_id = r.json()["id"]

    # 创建目标账号
    r = client.post(
        "/accounts",
        json={
            "asset_id": asset_id,
            "username": "root",
            "auth_type": "password",
            "credential": "supersecret",
        },
        headers=headers,
    )
    account_id = r.json()["id"]

    # 创建 operator 用户
    r = client.post(
        "/users",
        params={
            "username": "alice",
            "password": "alice123",
            "role": "operator",
        },
        headers=headers,
    )
    alice_id = r.json()["user_id"]

    # 授权
    r = client.post(
        "/grants",
        json={
            "user_id": alice_id,
            "asset_id": asset_id,
            "account_id": account_id,
            "can_shell": True,
            "valid_days": 7,
        },
        headers=headers,
    )
    assert r.status_code == 200

    # alice 登录并查看资产
    r = client.post(
        "/auth/login",
        json={"username": "alice", "password": "alice123"},
    )
    alice_token = r.json()["access_token"]
    alice_headers = {"Authorization": f"Bearer {alice_token}"}

    r = client.get("/assets/my", headers=alice_headers)
    assets = r.json()
    assert len(assets) == 1
    assert assets[0]["name"] == "web-01"


def test_command_policy_blocks_rm_rf():
    from app.policy import CommandAuditor
    from app.db import get_session

    token = _bootstrap_admin()
    headers = {"Authorization": f"Bearer {token}"}

    # 创建一个会话用于审计
    r = client.post(
        "/assets",
        json={"name": "db-01", "hostname": "10.0.0.20"},
        headers=headers,
    )
    asset_id = r.json()["id"]

    r = client.post(
        "/accounts",
        json={
            "asset_id": asset_id,
            "username": "root",
            "credential": "x",
        },
        headers=headers,
    )
    account_id = r.json()["id"]

    r = client.post(
        "/users",
        params={"username": "bob", "password": "bob123", "role": "operator"},
        headers=headers,
    )
    bob_id = r.json()["user_id"]

    client.post(
        "/grants",
        json={
            "user_id": bob_id,
            "asset_id": asset_id,
            "account_id": account_id,
        },
        headers=headers,
    )

    r = client.post(
        "/auth/login",
        json={"username": "bob", "password": "bob123"},
    )
    bob_headers = {"Authorization": f"Bearer {r.json()['access_token']}"}

    r = client.post(
        f"/sessions/start?asset_id={asset_id}",
        headers=bob_headers,
    )
    session_id = r.json()["session_id"]

    # 直接构造 auditor 做单元测试
    from app.db import engine
    from sqlmodel import Session as DBSession

    with DBSession(engine) as db:
        auditor = CommandAuditor(db, session_id, bob_id, asset_id)
        with pytest.raises(PermissionError):
            auditor.feed(b"rm -rf /\n")


def test_mfa_required_when_enabled():
    token = _bootstrap_admin()
    headers = {"Authorization": f"Bearer {token}"}

    r = client.post(
        "/users",
        params={
            "username": "mfa_user",
            "password": "mfa123",
            "role": "operator",
            "enable_mfa": True,
        },
        headers=headers,
    )
    assert "totp_secret" in r.json()

    r = client.post(
        "/auth/login",
        json={"username": "mfa_user", "password": "mfa123"},
    )
    assert r.status_code == 401
    assert "MFA" in r.json()["detail"]

pytest -v

FROM python:3.12-slim

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc libffi-dev libssl-dev && \
    rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

export BASTION_MASTER_KEY="$(python -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())')"
docker build -t enterprise-bastion .
docker run -p 8000:8000 -e BASTION_MASTER_KEY=$BASTION_MASTER_KEY \
  -v $(pwd)/recordings:/app/recordings enterprise-bastion

http://127.0.0.1:8000/docs

统一入口 + 身份认证 + 权限控制 + 全量审计 + 会话管控 + 合规报表

协议代理 -> 凭证托管 -> 授权引擎 -> 命令审计 -> 会话录制 -> 检索回放

enterprise-bastion/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── db.py
│   ├── models.py
│   ├── crypto.py
│   ├── auth.py
│   ├── rbac.py
│   ├── assets.py
│   ├── ssh_proxy.py
│   ├── audit.py
│   ├── policy.py
│   ├── recorder.py
│   └── playback.py
├── tests/
│   └── test_bastion.py
├── requirements.txt
└── Dockerfile

pip install -r requirements.txt
pytest -v
uvicorn app.main:app --reload