Phase 1 complete: Repository implementations + Service layer + API routes + Deploy docs
- All 11 Repository Protocol interfaces (Admin, Setting, AuditLog, PushSchedule, PushRetryJob, PasswordPreset + existing 5) - All 11 async SQLAlchemy Repository implementations - 4 Application Services (ServerService, ScriptService, AuthService, SyncService) - FastAPI DI wiring in dependencies.py (repos → services) - 6 API route modules (servers, auth, agent, scripts, settings/schedules/presets/audit/retries) - Agent /api/exec endpoint for remote shell command execution - Batch push engine with asyncio.Semaphore(10) concurrency control - TOTP authentication + brute-force protection (5 failures / 15 min lockout) - DB credential management with $DB_* variable substitution - Ubuntu deployment guide (docs/DEPLOY.md) with dependency fix script Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name
@@ -0,0 +1,493 @@
|
||||
# Nexus 部署指南 — Ubuntu 20.04 / 22.04 / 24.04
|
||||
|
||||
> **项目**: Nexus — 服务器运维管理平台
|
||||
> **版本**: 6.0.0
|
||||
> **技术栈**: Python 3.12+ / FastAPI / Async SQLAlchemy / MySQL 8.4 / Redis 7 / PHP 8.2+ / Nginx
|
||||
> **更新日期**: 2026-05-20
|
||||
|
||||
---
|
||||
|
||||
## 1. 系统依赖修复（⚠️ 首先运行）
|
||||
|
||||
Ubuntu 服务器部署时经常缺少编译和运行依赖。**在安装任何服务之前，先运行以下命令修复所有基础依赖：**
|
||||
|
||||
### 1.1 一键修复脚本
|
||||
|
||||
```bash
|
||||
#!/bin/bash
|
||||
# Nexus — Ubuntu 系统依赖一键修复脚本
|
||||
# 适用于 Ubuntu 20.04 / 22.04 / 24.04
|
||||
# 用法: sudo bash fix-deps.sh
|
||||
|
||||
set -e
|
||||
|
||||
echo "===== Nexus 系统依赖修复 ====="
|
||||
|
||||
# 更新软件源
|
||||
apt update
|
||||
|
||||
# ── 核心编译工具 ──
|
||||
apt install -y \
|
||||
build-essential \
|
||||
cmake \
|
||||
autoconf \
|
||||
automake \
|
||||
libtool \
|
||||
pkg-config \
|
||||
gcc \
|
||||
g++ \
|
||||
make \
|
||||
git \
|
||||
curl \
|
||||
wget \
|
||||
unzip \
|
||||
ca-certificates \
|
||||
gnupg
|
||||
|
||||
# ── 通用开发库（所有组件共享）──
|
||||
apt install -y \
|
||||
libssl-dev \
|
||||
libxml2-dev \
|
||||
libcurl4-openssl-dev \
|
||||
zlib1g-dev \
|
||||
libpcre3-dev \
|
||||
libpcre2-dev \
|
||||
libbz2-dev \
|
||||
libzip-dev \
|
||||
liblz4-dev \
|
||||
libzstd-dev \
|
||||
liblzma-dev \
|
||||
libffi-dev \
|
||||
libncurses5-dev \
|
||||
libncursesw5-dev \
|
||||
libreadline-dev \
|
||||
libsqlite3-dev \
|
||||
libicu-dev \
|
||||
libbrotli-dev
|
||||
|
||||
# ── Nginx 专用依赖 ──
|
||||
apt install -y \
|
||||
libgd-dev \
|
||||
libgeoip-dev \
|
||||
libxslt1-dev
|
||||
|
||||
# ── PHP 8.x 专用依赖 ──
|
||||
apt install -y \
|
||||
libonig-dev \
|
||||
libsodium-dev \
|
||||
libargon2-dev \
|
||||
libjpeg-dev \
|
||||
libpng-dev \
|
||||
libfreetype6-dev \
|
||||
libwebp-dev \
|
||||
libgmp-dev \
|
||||
libtidy-dev \
|
||||
libpq-dev \
|
||||
default-libmysqlclient-dev
|
||||
|
||||
# ── MySQL 8.4 专用依赖 ──
|
||||
apt install -y \
|
||||
bison \
|
||||
gperf \
|
||||
libevent-dev \
|
||||
libnuma-dev \
|
||||
libaio-dev \
|
||||
protobuf-compiler \
|
||||
libprotobuf-dev
|
||||
|
||||
# ── Python 3.12+ 专用依赖 ──
|
||||
apt install -y \
|
||||
libgdbm-dev \
|
||||
libgdbm-compat-dev \
|
||||
libmpdec-dev \
|
||||
uuid-dev
|
||||
|
||||
# ── Redis 7 专用依赖 ──
|
||||
apt install -y \
|
||||
libjemalloc-dev
|
||||
|
||||
echo "===== 依赖修复完成 ====="
|
||||
```
|
||||
|
||||
### 1.2 手动逐项安装（如果脚本不适用）
|
||||
|
||||
按组件分类的依赖列表：
|
||||
|
||||
#### 通用依赖（所有组件都需要）
|
||||
```bash
|
||||
sudo apt install -y build-essential cmake autoconf automake libtool pkg-config \
|
||||
gcc g++ make git curl wget unzip ca-certificates gnupg \
|
||||
libssl-dev libxml2-dev libcurl4-openssl-dev zlib1g-dev \
|
||||
libpcre3-dev libpcre2-dev libbz2-dev libzip-dev \
|
||||
liblz4-dev libzstd-dev liblzma-dev libffi-dev \
|
||||
libncurses5-dev libncursesw5-dev libreadline-dev \
|
||||
libsqlite3-dev libicu-dev libbrotli-dev
|
||||
```
|
||||
|
||||
#### Nginx 专用
|
||||
```bash
|
||||
sudo apt install -y libgd-dev libgeoip-dev libxslt1-dev
|
||||
```
|
||||
|
||||
#### PHP 8.x 专用
|
||||
```bash
|
||||
sudo apt install -y libonig-dev libsodium-dev libargon2-dev \
|
||||
libjpeg-dev libpng-dev libfreetype6-dev libwebp-dev \
|
||||
libgmp-dev libtidy-dev libpq-dev default-libmysqlclient-dev
|
||||
```
|
||||
|
||||
#### MySQL 8.4 专用
|
||||
```bash
|
||||
sudo apt install -y bison gperf libevent-dev libnuma-dev \
|
||||
libaio-dev protobuf-compiler libprotobuf-dev
|
||||
```
|
||||
|
||||
#### Python 3.12+ 专用
|
||||
```bash
|
||||
sudo apt install -y libgdbm-dev libgdbm-compat-dev \
|
||||
libmpdec-dev uuid-dev
|
||||
```
|
||||
|
||||
#### Redis 专用
|
||||
```bash
|
||||
sudo apt install -y libjemalloc-dev
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 2. 服务安装
|
||||
|
||||
### 2.1 Nginx
|
||||
|
||||
```bash
|
||||
# 方式 A: apt 直接安装（推荐）
|
||||
sudo apt install -y nginx
|
||||
|
||||
# 方式 B: 编译安装（需要更多自定义模块时）
|
||||
# 编译依赖已在 1.1 中安装
|
||||
wget http://nginx.org/download/nginx-1.26.2.tar.gz
|
||||
tar -xzf nginx-1.26.2.tar.gz
|
||||
cd nginx-1.26.2
|
||||
./configure --prefix=/usr/local/nginx \
|
||||
--with-http_ssl_module \
|
||||
--with-http_v2_module \
|
||||
--with-http_realip_module \
|
||||
--with-http_gzip_static_module \
|
||||
--with-http_brotli_module \
|
||||
--with-stream \
|
||||
--with-stream_ssl_module
|
||||
make && sudo make install
|
||||
```
|
||||
|
||||
### 2.2 MySQL 8.4 LTS
|
||||
|
||||
```bash
|
||||
# 方式 A: apt 安装（推荐 — MySQL 官方 APT 源）
|
||||
wget https://dev.mysql.com/get/mysql-apt-config_0.8.33-1ubuntu_all.deb
|
||||
sudo dpkg -i mysql-apt-config_0.8.33-1ubuntu_all.deb
|
||||
# 在配置界面选择 mysql-8.4-lts
|
||||
sudo apt update
|
||||
sudo apt install -y mysql-server mysql-client
|
||||
|
||||
# 安全初始化
|
||||
sudo mysql_secure_installation
|
||||
|
||||
# 方式 B: 宝塔面板一键安装
|
||||
# 在宝塔面板 → 软件商店 → MySQL 8.4 → 安装
|
||||
```
|
||||
|
||||
### 2.3 Redis 7
|
||||
|
||||
```bash
|
||||
# 方式 A: apt 安装
|
||||
sudo apt install -y redis-server
|
||||
|
||||
# 方式 B: 编译安装（最新版本）
|
||||
wget https://github.com/redis/redis/archive/refs/tags/7.4.2.tar.gz
|
||||
tar -xzf 7.4.2.tar.gz
|
||||
cd redis-7.4.2
|
||||
make MALLOC=jemalloc
|
||||
sudo make install
|
||||
|
||||
# 启动
|
||||
redis-server --daemonize yes
|
||||
```
|
||||
|
||||
### 2.4 PHP 8.2+
|
||||
|
||||
```bash
|
||||
# 方式 A: apt 安装（Ubuntu 22.04+ 默认有 PHP 8.1）
|
||||
sudo apt install -y php8.2 php8.2-fpm php8.2-cli \
|
||||
php8.2-mysql php8.2-redis php8.2-gd php8.2-mbstring \
|
||||
php8.2-xml php8.2-zip php8.2-curl php8.2-bz2 \
|
||||
php8.2-intl php8.2-opcache php8.2-json php8.2-readline
|
||||
|
||||
# 如果需要 PHP 8.3（更新版本）:
|
||||
sudo add-apt-repository -y ppa:ondrej/php
|
||||
sudo apt update
|
||||
sudo apt install -y php8.3 php8.3-fpm php8.3-cli \
|
||||
php8.3-mysql php8.3-redis php8.3-gd php8.3-mbstring \
|
||||
php8.3-xml php8.3-zip php8.3-curl php8.3-bz2 \
|
||||
php8.3-intl php8.3-opcache
|
||||
|
||||
# 方式 B: 宝塔面板安装
|
||||
# 宝塔面板 → 软件商店 → PHP 8.2/8.3 → 安装
|
||||
```
|
||||
|
||||
### 2.5 Python 3.12+
|
||||
|
||||
```bash
|
||||
# 方式 A: apt 安装（Ubuntu 24.04 默认有 Python 3.12）
|
||||
sudo apt install -y python3.12 python3.12-dev python3.12-venv \
|
||||
python3-pip python3-venv
|
||||
|
||||
# Ubuntu 22.04 需要添加 deadsnakes PPA:
|
||||
sudo add-apt-repository -y ppa:deadsnakes/ppa
|
||||
sudo apt update
|
||||
sudo apt install -y python3.12 python3.12-dev python3.12-venv
|
||||
|
||||
# 方式 B: 编译安装
|
||||
wget https://www.python.org/ftp/python/3.12.7/Python-3.12.7.tar.xz
|
||||
tar -xf Python-3.12.7.tar.xz
|
||||
cd Python-3.12.7
|
||||
./configure --enable-optimizations --with-ensurepip=install
|
||||
make -j$(nproc)
|
||||
sudo make altinstall # altinstall 防止覆盖系统 python3
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 3. Nexus 项目部署
|
||||
|
||||
### 3.1 克隆仓库
|
||||
|
||||
```bash
|
||||
cd /opt
|
||||
git clone http://66.154.115.8:3000/admin/Nexus.git
|
||||
cd Nexus
|
||||
```
|
||||
|
||||
### 3.2 Python 环境配置
|
||||
|
||||
```bash
|
||||
# 创建虚拟环境
|
||||
python3.12 -m venv venv
|
||||
source venv/bin/activate
|
||||
|
||||
# 安装 Python 依赖
|
||||
pip install -r requirements.txt
|
||||
|
||||
# 如果 pip 安装遇到编译错误（缺少依赖），回到步骤 1 修复依赖后重新安装
|
||||
```
|
||||
|
||||
### 3.3 配置文件
|
||||
|
||||
```bash
|
||||
# 复制配置模板
|
||||
cp .env.example .env
|
||||
|
||||
# 编辑 .env 文件
|
||||
nano .env
|
||||
```
|
||||
|
||||
`.env` 必填项：
|
||||
|
||||
```ini
|
||||
# ⚠️ SECRET_KEY 必须设置，否则 Nexus 无法启动
|
||||
NEXUS_SECRET_KEY=<生成一个随机密钥: python3 -c "import secrets; print(secrets.token_urlsafe(32))">
|
||||
|
||||
# 数据库连接（aiomysql 驱动）
|
||||
NEXUS_DATABASE_URL=mysql+aiomysql://nexus:your_password@127.0.0.1:3306/nexus
|
||||
|
||||
# Redis
|
||||
NEXUS_REDIS_URL=redis://127.0.0.1:6379/0
|
||||
|
||||
# API Key（Agent 认证用）
|
||||
NEXUS_API_KEY=<生成: python3 -c "import secrets; print(secrets.token_urlsafe(24))">
|
||||
|
||||
# 加密密钥（可选，默认从 SECRET_KEY 派生）
|
||||
NEXUS_ENCRYPTION_KEY=
|
||||
```
|
||||
|
||||
### 3.4 创建 MySQL 数据库
|
||||
|
||||
```bash
|
||||
sudo mysql -u root -p
|
||||
|
||||
# 在 MySQL 中执行:
|
||||
CREATE DATABASE nexus CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
|
||||
CREATE USER 'nexus'@'localhost' IDENTIFIED BY 'your_password';
|
||||
GRANT ALL PRIVILEGES ON nexus.* TO 'nexus'@'localhost';
|
||||
FLUSH PRIVILEGES;
|
||||
EXIT;
|
||||
```
|
||||
|
||||
> **注意**: Nexus 启动时会自动创建所有数据表（`Base.metadata.create_all`），无需手动建表。
|
||||
|
||||
### 3.5 启动 Nexus
|
||||
|
||||
```bash
|
||||
# 方式 A: 直接运行（开发/调试）
|
||||
source venv/bin/activate
|
||||
uvicorn server.main:app --host 0.0.0.0 --port 8600
|
||||
|
||||
# 方式 B: Supervisor 守护进程（生产环境推荐）
|
||||
sudo apt install -y supervisor
|
||||
|
||||
# 创建 supervisor 配置
|
||||
sudo nano /etc/supervisor/conf.d/nexus.conf
|
||||
```
|
||||
|
||||
`nexus.conf` 内容：
|
||||
|
||||
```ini
|
||||
[program:nexus]
|
||||
command=/opt/Nexus/venv/bin/uvicorn server.main:app --host 0.0.0.0 --port 8600 --workers 4
|
||||
directory=/opt/Nexus
|
||||
user=root
|
||||
autostart=true
|
||||
autorestart=true
|
||||
stderr_logfile=/var/log/nexus/error.log
|
||||
stdout_logfile=/var/log/nexus/access.log
|
||||
environment=NEXUS_SECRET_KEY="your_secret_key",NEXUS_DATABASE_URL="mysql+aiomysql://nexus:password@127.0.0.1:3306/nexus"
|
||||
```
|
||||
|
||||
```bash
|
||||
sudo supervisorctl reread
|
||||
sudo supervisorctl update
|
||||
sudo supervisorctl start nexus
|
||||
```
|
||||
|
||||
### 3.6 Nginx 反向代理配置
|
||||
|
||||
```bash
|
||||
sudo nano /etc/nginx/sites-available/nexus
|
||||
```
|
||||
|
||||
```nginx
|
||||
server {
|
||||
listen 80;
|
||||
server_name your-domain.com;
|
||||
|
||||
# 强制 HTTPS
|
||||
return 301 https://$server_name$request_uri;
|
||||
}
|
||||
|
||||
server {
|
||||
listen 443 ssl http2;
|
||||
server_name your-domain.com;
|
||||
|
||||
# SSL证书（宝塔面板一键配置 Let's Encrypt）
|
||||
ssl_certificate /etc/letsencrypt/live/your-domain.com/fullchain.pem;
|
||||
ssl_certificate_key /etc/letsencrypt/live/your-domain.com/privkey.pem;
|
||||
|
||||
# Nexus API
|
||||
location /api/ {
|
||||
proxy_pass http://127.0.0.1:8600;
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_set_header X-Forwarded-Proto $scheme;
|
||||
}
|
||||
|
||||
# Nexus WebSocket
|
||||
location /ws/ {
|
||||
proxy_pass http://127.0.0.1:8600;
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Upgrade $http_upgrade;
|
||||
proxy_set_header Connection "upgrade";
|
||||
}
|
||||
|
||||
# PHP 页面（前端）
|
||||
location / {
|
||||
root /opt/Nexus/web;
|
||||
index index.php;
|
||||
try_files $uri $uri/ /index.php?$query_string;
|
||||
|
||||
location ~ \.php$ {
|
||||
fastcgi_pass unix:/run/php/php8.2-fpm.sock;
|
||||
fastcgi_param SCRIPT_FILENAME $document_root$fastcgi_script_name;
|
||||
include fastcgi_params;
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
```bash
|
||||
sudo ln -s /etc/nginx/sites-available/nexus /etc/nginx/sites-enabled/
|
||||
sudo nginx -t
|
||||
sudo systemctl reload nginx
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 4. 依赖冲突排查
|
||||
|
||||
### 4.1 常见问题
|
||||
|
||||
| 问题 | 原因 | 解决方案 |
|
||||
|------|------|----------|
|
||||
| `pip install` 报错 `libbrotli` | 缺少 `libbrotli-dev` | `sudo apt install -y libbrotli-dev` |
|
||||
| `pip install` 报错 `libssl` | 缺少 `libssl-dev` | `sudo apt install -y libssl-dev` |
|
||||
| `pip install` 报错 `libcurl` | 缺少 `libcurl4-openssl-dev` | `sudo apt install -y libcurl4-openssl-dev` |
|
||||
| PHP `mbstring` 编译失败 | 缺少 `libonig-dev` | `sudo apt install -y libonig-dev` |
|
||||
| PHP `xml/dom` 编译失败 | 缺少 `libxml2-dev` | `sudo apt install -y libxml2-dev` |
|
||||
| PHP `zip` 编译失败 | 缺少 `libzip-dev` | `sudo apt install -y libzip-dev` |
|
||||
| `libcurl4-openssl-dev` 与 `libcurl4-nss-dev` 冲突 | 不能同时安装两个 curl dev 包 | `sudo apt remove libcurl4-nss-dev && sudo apt install libcurl4-openssl-dev` |
|
||||
| MySQL 编译报错缺少 `bison/gperf` | 缺少 MySQL 编译依赖 | `sudo apt install -y bison gperf` |
|
||||
| Python `decimal` 模块报错 | 缺少 `libmpdec-dev`（Python 3.12+ 新依赖） | `sudo apt install -y libmpdec-dev` |
|
||||
|
||||
### 4.2 快速检查缺少的依赖
|
||||
|
||||
```bash
|
||||
# 检查某个库是否已安装
|
||||
dpkg -l | grep libbrotli-dev
|
||||
apt-cache search libbrotli
|
||||
|
||||
# 查看某个 pip 包需要哪些系统依赖
|
||||
pip show <package-name>
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 5. 版本兼容性参考
|
||||
|
||||
| 组件 | Ubuntu 20.04 | Ubuntu 22.04 | Ubuntu 24.04 | 说明 |
|
||||
|------|-------------|-------------|-------------|------|
|
||||
| Python | 3.8 (系统) → 需 deadsnakes PPA 安装 3.12 | 3.10 (系统) → deadsnakes PPA 安装 3.12 | **3.12 (系统)** | 推荐 3.12+ |
|
||||
| MySQL | 8.0 (系统) → 可手动安装 8.4 | 8.0 (系统) → 可手动安装 8.4 | **8.4 可用** | 推荐 8.4 LTS |
|
||||
| Redis | 6.x (系统) → 需编译 7.x | 6.x→7.x (系统) | **7.x (系统)** | 推荐 7.0+ |
|
||||
| PHP | 7.4 (系统) → 需 ondrej PPA 安装 8.2 | 8.1 (系统) → 需 PPA 安装 8.2/8.3 | **8.3 (系统)** | 推荐 8.2+ |
|
||||
| Nginx | 1.18 (系统) → 可编译最新 | 1.22 (系统) → 可编译最新 | **1.24 (系统)** | 推荐 1.24+ |
|
||||
|
||||
---
|
||||
|
||||
## 6. 宝塔面板部署（推荐新手）
|
||||
|
||||
如果使用宝塔面板，大部分依赖和软件安装可以一键完成：
|
||||
|
||||
1. 安装宝塔面板: `wget -O install.sh https://download.bt.cn/install/install-ubuntu_6.0.sh && sudo bash install.sh ed848235`
|
||||
2. 宝塔面板 → 软件商店 → 安装: Nginx / MySQL 8.4 / Redis 7 / PHP 8.2
|
||||
3. 创建网站 → 设置根目录为 `/opt/Nexus/web`
|
||||
4. 配置反向代理 → 将 `/api/` 代理到 `127.0.0.1:8600`
|
||||
5. SSL → Let's Encrypt 一键申请
|
||||
|
||||
> **宝塔面板会自动安装大部分系统依赖**，但 Python pip 依赖仍需手动处理。如果 pip 安装报错，回到步骤 1 安装缺失的 `-dev` 包。
|
||||
|
||||
---
|
||||
|
||||
## 7. 验证部署
|
||||
|
||||
```bash
|
||||
# 检查 Nexus API 是否运行
|
||||
curl http://127.0.0.1:8600/docs # 应返回 OpenAPI 文档页面
|
||||
|
||||
# 检查数据库连接
|
||||
curl http://127.0.0.1:8600/api/settings/ # 应返回设置列表（初始为空）
|
||||
|
||||
# 检查 Agent 心跳端点
|
||||
curl -X POST http://127.0.0.1:8600/api/agent/heartbeat \
|
||||
-H "X-API-Key: your_api_key" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"server_id": 1, "is_online": true}'
|
||||
``
|
||||
@@ -0,0 +1,111 @@
|
||||
"""Nexus — Agent API Routes (Heartbeat receiver + exec endpoint)
|
||||
Presentation layer — Agent servers call these endpoints to report status and receive commands.
|
||||
"""
|
||||
|
||||
import json
|
||||
from fastapi import APIRouter, Depends, Header, HTTPException
|
||||
|
||||
from server.api.dependencies import get_server_service
|
||||
from server.application.services.server_service import ServerService
|
||||
from server.config import settings
|
||||
|
||||
router = APIRouter(prefix="/api/agent", tags=["agent"])
|
||||
|
||||
|
||||
def _verify_api_key(x_api_key: str = Header(...)):
|
||||
"""Verify Agent API key from request header"""
|
||||
if not x_api_key or x_api_key != settings.API_KEY:
|
||||
raise HTTPException(status_code=401, detail="Invalid API key")
|
||||
return x_api_key
|
||||
|
||||
|
||||
@router.post("/heartbeat", response_model=dict)
|
||||
async def receive_heartbeat(
|
||||
payload: dict,
|
||||
api_key: str = Depends(_verify_api_key),
|
||||
service: ServerService = Depends(get_server_service),
|
||||
):
|
||||
"""Receive Agent heartbeat data
|
||||
|
||||
Body: {
|
||||
"server_id": 1, // or "server_name": "web-01"
|
||||
"is_online": true,
|
||||
"system_info": {"cpu": 45, "mem": 60, "disk": 70, ...},
|
||||
"agent_version": "1.2.0"
|
||||
}
|
||||
"""
|
||||
server_id = payload.get("server_id")
|
||||
if not server_id:
|
||||
# Lookup by name if ID not provided
|
||||
server_name = payload.get("server_name")
|
||||
if server_name:
|
||||
# TODO: Add get_by_name to ServerRepository
|
||||
raise HTTPException(status_code=400, detail="server_id required (name lookup not yet implemented)")
|
||||
raise HTTPException(status_code=400, detail="server_id required")
|
||||
|
||||
await service.update_heartbeat(server_id, payload)
|
||||
return {"status": "ok", "server_id": server_id}
|
||||
|
||||
|
||||
@router.post("/exec", response_model=dict)
|
||||
async def agent_exec(
|
||||
payload: dict,
|
||||
api_key: str = Depends(_verify_api_key),
|
||||
):
|
||||
"""Execute a shell command on this Agent server
|
||||
|
||||
This endpoint is called by Nexus to dispatch commands to remote Agents.
|
||||
Each Agent server runs its own instance of this endpoint.
|
||||
|
||||
Body: {
|
||||
"command": "systemctl restart nginx",
|
||||
"timeout": 30, // seconds, default 30, max 300
|
||||
"sudo": false // whether to run with sudo
|
||||
}
|
||||
"""
|
||||
command = payload.get("command")
|
||||
if not command:
|
||||
raise HTTPException(status_code=400, detail="command required")
|
||||
|
||||
timeout = min(payload.get("timeout", 30), 300) # Max 5 minutes
|
||||
use_sudo = payload.get("sudo", False)
|
||||
|
||||
if use_sudo:
|
||||
command = f"sudo {command}"
|
||||
|
||||
# Execute command locally
|
||||
import asyncio
|
||||
try:
|
||||
proc = await asyncio.create_subprocess_shell(
|
||||
command,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||||
exit_code = proc.returncode or 0
|
||||
|
||||
# Truncate output to 10KB
|
||||
stdout_text = stdout.decode()[:10000]
|
||||
stderr_text = stderr.decode()[:10000]
|
||||
|
||||
return {
|
||||
"status": "success" if exit_code == 0 else "failed",
|
||||
"stdout": stdout_text,
|
||||
"stderr": stderr_text,
|
||||
"exit_code": exit_code,
|
||||
}
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
return {
|
||||
"status": "timeout",
|
||||
"stdout": "",
|
||||
"stderr": f"Command timed out after {timeout}s",
|
||||
"exit_code": -1,
|
||||
}
|
||||
except Exception as e:
|
||||
return {
|
||||
"status": "error",
|
||||
"stdout": "",
|
||||
"stderr": str(e),
|
||||
"exit_code": -1,
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Nexus — Auth API Routes (Login / TOTP / Session management)
|
||||
Presentation layer — receives HTTP requests, delegates to AuthService.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
|
||||
from server.api.dependencies import get_auth_service
|
||||
from server.application.services.auth_service import AuthService
|
||||
|
||||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||||
|
||||
|
||||
@router.post("/login", response_model=dict)
|
||||
async def login(
|
||||
request: Request,
|
||||
payload: dict,
|
||||
service: AuthService = Depends(get_auth_service),
|
||||
):
|
||||
"""Authenticate admin user (password + optional TOTP)
|
||||
|
||||
Body: {
|
||||
"username": "admin",
|
||||
"password": "...",
|
||||
"totp_code": "123456" // Optional, required if TOTP enabled
|
||||
}
|
||||
"""
|
||||
ip_address = request.client.host if request.client else "unknown"
|
||||
result = await service.login(
|
||||
username=payload.get("username", ""),
|
||||
password=payload.get("password", ""),
|
||||
ip_address=ip_address,
|
||||
totp_code=payload.get("totp_code"),
|
||||
)
|
||||
if not result["success"]:
|
||||
status_code = 401
|
||||
if result.get("reason") == "account_locked":
|
||||
status_code = 429
|
||||
raise HTTPException(status_code=status_code, detail=result.get("message", "Login failed"))
|
||||
return result
|
||||
|
||||
|
||||
@router.post("/totp/setup", response_model=dict)
|
||||
async def setup_totp(
|
||||
payload: dict,
|
||||
service: AuthService = Depends(get_auth_service),
|
||||
):
|
||||
"""Generate TOTP secret for admin user (requires valid session token)
|
||||
|
||||
Body: {"admin_id": 1}
|
||||
"""
|
||||
# TODO: Verify session token from Authorization header
|
||||
admin_id = payload.get("admin_id")
|
||||
if not admin_id:
|
||||
raise HTTPException(status_code=400, detail="admin_id required")
|
||||
result = await service.setup_totp(admin_id)
|
||||
if not result["success"]:
|
||||
raise HTTPException(status_code=400, detail=result.get("reason", "Setup failed"))
|
||||
return result
|
||||
|
||||
|
||||
@router.post("/totp/enable", response_model=dict)
|
||||
async def enable_totp(
|
||||
payload: dict,
|
||||
service: AuthService = Depends(get_auth_service),
|
||||
):
|
||||
"""Enable TOTP after verification (requires valid session token)
|
||||
|
||||
Body: {"admin_id": 1, "totp_code": "123456"}
|
||||
"""
|
||||
admin_id = payload.get("admin_id")
|
||||
totp_code = payload.get("totp_code")
|
||||
if not admin_id or not totp_code:
|
||||
raise HTTPException(status_code=400, detail="admin_id and totp_code required")
|
||||
result = await service.enable_totp(admin_id, totp_code)
|
||||
if not result["success"]:
|
||||
raise HTTPException(status_code=400, detail=result.get("message", "Enable failed"))
|
||||
return result
|
||||
|
||||
|
||||
@router.post("/totp/disable", response_model=dict)
|
||||
async def disable_totp(
|
||||
payload: dict,
|
||||
service: AuthService = Depends(get_auth_service),
|
||||
):
|
||||
"""Disable TOTP for admin user (requires valid session token)
|
||||
|
||||
Body: {"admin_id": 1}
|
||||
"""
|
||||
admin_id = payload.get("admin_id")
|
||||
if not admin_id:
|
||||
raise HTTPException(status_code=400, detail="admin_id required")
|
||||
result = await service.disable_totp(admin_id)
|
||||
return result
|
||||
@@ -0,0 +1,112 @@
|
||||
"""Nexus — FastAPI Dependency Injection
|
||||
Wires Repository implementations → Service instances → API route handlers.
|
||||
All DI occurs here; routes never import infrastructure directly.
|
||||
"""
|
||||
|
||||
from typing import AsyncGenerator
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.infrastructure.database.session import get_async_session, AsyncSessionLocal
|
||||
from server.infrastructure.database.server_repo import ServerRepositoryImpl
|
||||
from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
|
||||
from server.infrastructure.database.script_repo import ScriptRepositoryImpl, ScriptExecutionRepositoryImpl
|
||||
from server.infrastructure.database.db_credential_repo import DbCredentialRepositoryImpl
|
||||
from server.infrastructure.database.admin_repo import AdminRepositoryImpl, LoginAttemptRepositoryImpl
|
||||
from server.infrastructure.database.setting_repo import SettingRepositoryImpl
|
||||
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
|
||||
from server.infrastructure.database.push_schedule_repo import PushScheduleRepositoryImpl, PushRetryJobRepositoryImpl
|
||||
from server.infrastructure.database.password_preset_repo import PasswordPresetRepositoryImpl
|
||||
|
||||
from server.application.services.server_service import ServerService
|
||||
from server.application.services.script_service import ScriptService
|
||||
from server.application.services.auth_service import AuthService
|
||||
from server.application.services.sync_service import SyncService
|
||||
|
||||
|
||||
async def get_db() -> AsyncGenerator[AsyncSession, None]:
|
||||
"""Yield an async database session (for use as FastAPI Depends)"""
|
||||
session = AsyncSessionLocal()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
await session.close()
|
||||
|
||||
|
||||
# ── Repository Factories ──
|
||||
|
||||
def _server_repo(session: AsyncSession) -> ServerRepositoryImpl:
|
||||
return ServerRepositoryImpl(session)
|
||||
|
||||
def _sync_log_repo(session: AsyncSession) -> SyncLogRepositoryImpl:
|
||||
return SyncLogRepositoryImpl(session)
|
||||
|
||||
def _script_repo(session: AsyncSession) -> ScriptRepositoryImpl:
|
||||
return ScriptRepositoryImpl(session)
|
||||
|
||||
def _execution_repo(session: AsyncSession) -> ScriptExecutionRepositoryImpl:
|
||||
return ScriptExecutionRepositoryImpl(session)
|
||||
|
||||
def _credential_repo(session: AsyncSession) -> DbCredentialRepositoryImpl:
|
||||
return DbCredentialRepositoryImpl(session)
|
||||
|
||||
def _admin_repo(session: AsyncSession) -> AdminRepositoryImpl:
|
||||
return AdminRepositoryImpl(session)
|
||||
|
||||
def _attempt_repo(session: AsyncSession) -> LoginAttemptRepositoryImpl:
|
||||
return LoginAttemptRepositoryImpl(session)
|
||||
|
||||
def _setting_repo(session: AsyncSession) -> SettingRepositoryImpl:
|
||||
return SettingRepositoryImpl(session)
|
||||
|
||||
def _audit_repo(session: AsyncSession) -> AuditLogRepositoryImpl:
|
||||
return AuditLogRepositoryImpl(session)
|
||||
|
||||
def _schedule_repo(session: AsyncSession) -> PushScheduleRepositoryImpl:
|
||||
return PushScheduleRepositoryImpl(session)
|
||||
|
||||
def _retry_repo(session: AsyncSession) -> PushRetryJobRepositoryImpl:
|
||||
return PushRetryJobRepositoryImpl(session)
|
||||
|
||||
def _preset_repo(session: AsyncSession) -> PasswordPresetRepositoryImpl:
|
||||
return PasswordPresetRepositoryImpl(session)
|
||||
|
||||
|
||||
# ── Service Factories (compose repositories) ──
|
||||
|
||||
async def get_server_service() -> ServerService:
|
||||
session = AsyncSessionLocal()
|
||||
return ServerService(
|
||||
server_repo=_server_repo(session),
|
||||
sync_log_repo=_sync_log_repo(session),
|
||||
)
|
||||
|
||||
|
||||
async def get_script_service() -> ScriptService:
|
||||
session = AsyncSessionLocal()
|
||||
return ScriptService(
|
||||
script_repo=_script_repo(session),
|
||||
execution_repo=_execution_repo(session),
|
||||
credential_repo=_credential_repo(session),
|
||||
server_repo=_server_repo(session),
|
||||
audit_repo=_audit_repo(session),
|
||||
)
|
||||
|
||||
|
||||
async def get_auth_service() -> AuthService:
|
||||
session = AsyncSessionLocal()
|
||||
return AuthService(
|
||||
admin_repo=_admin_repo(session),
|
||||
attempt_repo=_attempt_repo(session),
|
||||
audit_repo=_audit_repo(session),
|
||||
)
|
||||
|
||||
|
||||
async def get_sync_service() -> SyncService:
|
||||
session = AsyncSessionLocal()
|
||||
return SyncService(
|
||||
server_repo=_server_repo(session),
|
||||
sync_log_repo=_sync_log_repo(session),
|
||||
audit_repo=_audit_repo(session),
|
||||
retry_repo=_retry_repo(session),
|
||||
)
|
||||
@@ -0,0 +1,217 @@
|
||||
"""Nexus — Scripts API Routes (Script library CRUD + command execution)
|
||||
Presentation layer — receives HTTP requests, delegates to ScriptService.
|
||||
"""
|
||||
|
||||
from typing import Optional, List
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
|
||||
from server.api.dependencies import get_script_service
|
||||
from server.application.services.script_service import ScriptService
|
||||
from server.domain.models import Script, DbCredential
|
||||
|
||||
router = APIRouter(prefix="/api/scripts", tags=["scripts"])
|
||||
|
||||
|
||||
# ── Script CRUD ──
|
||||
|
||||
@router.get("/", response_model=list)
|
||||
async def list_scripts(
|
||||
category: Optional[str] = Query(None, description="Filter by category"),
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""List all scripts, optionally filtered by category"""
|
||||
scripts = await service.list_scripts(category)
|
||||
return [_script_to_dict(s) for s in scripts]
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=dict)
|
||||
async def get_script(
|
||||
id: int,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Get a single script by ID"""
|
||||
script = await service.get_script(id)
|
||||
if not script:
|
||||
raise HTTPException(status_code=404, detail="Script not found")
|
||||
return _script_to_dict(script)
|
||||
|
||||
|
||||
@router.post("/", response_model=dict, status_code=201)
|
||||
async def create_script(
|
||||
payload: dict,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Create a new script
|
||||
|
||||
Body: {
|
||||
"name": "restart-nginx",
|
||||
"category": "ops",
|
||||
"content": "systemctl restart nginx",
|
||||
"description": "Restart nginx service",
|
||||
"created_by": "admin"
|
||||
}
|
||||
"""
|
||||
script = Script(**payload)
|
||||
created = await service.create_script(script)
|
||||
return _script_to_dict(created)
|
||||
|
||||
|
||||
@router.put("/{id}", response_model=dict)
|
||||
async def update_script(
|
||||
id: int,
|
||||
payload: dict,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Update an existing script"""
|
||||
script = await service.get_script(id)
|
||||
if not script:
|
||||
raise HTTPException(status_code=404, detail="Script not found")
|
||||
for key, value in payload.items():
|
||||
if hasattr(script, key) and key != "id":
|
||||
setattr(script, key, value)
|
||||
updated = await service.update_script(script)
|
||||
return _script_to_dict(updated)
|
||||
|
||||
|
||||
@router.delete("/{id}", status_code=204)
|
||||
async def delete_script(
|
||||
id: int,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Delete a script"""
|
||||
result = await service.delete_script(id)
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="Script not found")
|
||||
|
||||
|
||||
# ── Command Execution ──
|
||||
|
||||
@router.post("/exec", response_model=dict)
|
||||
async def execute_command(
|
||||
payload: dict,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Execute a shell command on multiple servers via Agent /api/exec
|
||||
|
||||
Body: {
|
||||
"command": "systemctl restart nginx", // or script content
|
||||
"server_ids": [1, 2, 3],
|
||||
"script_id": null, // Optional: link to saved script
|
||||
"credential_id": null, // Optional: DB credential for $DB_* vars
|
||||
"timeout": 30,
|
||||
"operator": "admin"
|
||||
}
|
||||
"""
|
||||
command = payload.get("command")
|
||||
server_ids = payload.get("server_ids", [])
|
||||
if not command or not server_ids:
|
||||
raise HTTPException(status_code=400, detail="command and server_ids required")
|
||||
|
||||
execution = await service.execute_command(
|
||||
command=command,
|
||||
server_ids=server_ids,
|
||||
script_id=payload.get("script_id"),
|
||||
credential_id=payload.get("credential_id"),
|
||||
timeout=payload.get("timeout", 30),
|
||||
operator=payload.get("operator", "admin"),
|
||||
)
|
||||
return _execution_to_dict(execution)
|
||||
|
||||
|
||||
@router.get("/executions/{id}", response_model=dict)
|
||||
async def get_execution(
|
||||
id: int,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Get execution result by ID"""
|
||||
execution = await service.get_execution(id)
|
||||
if not execution:
|
||||
raise HTTPException(status_code=404, detail="Execution not found")
|
||||
return _execution_to_dict(execution)
|
||||
|
||||
|
||||
# ── DB Credentials ──
|
||||
|
||||
@router.get("/credentials", response_model=list)
|
||||
async def list_credentials(
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""List all database credentials"""
|
||||
credentials = await service.list_credentials()
|
||||
return [_credential_to_dict(c) for c in credentials]
|
||||
|
||||
|
||||
@router.post("/credentials", response_model=dict, status_code=201)
|
||||
async def create_credential(
|
||||
payload: dict,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Create a database credential (password will be encrypted)
|
||||
|
||||
Body: {
|
||||
"name": "production-mysql",
|
||||
"db_type": "mysql",
|
||||
"host": "localhost",
|
||||
"port": 3306,
|
||||
"username": "root",
|
||||
"encrypted_password": "actual_password", // Will be encrypted before storage
|
||||
"database": "app_db"
|
||||
}
|
||||
"""
|
||||
credential = DbCredential(**payload)
|
||||
created = await service.create_credential(credential)
|
||||
return _credential_to_dict(created)
|
||||
|
||||
|
||||
@router.delete("/credentials/{id}", status_code=204)
|
||||
async def delete_credential(
|
||||
id: int,
|
||||
service: ScriptService = Depends(get_script_service),
|
||||
):
|
||||
"""Delete a database credential"""
|
||||
result = await service.delete_credential(id)
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="Credential not found")
|
||||
|
||||
|
||||
# ── Helper functions ──
|
||||
|
||||
def _script_to_dict(script: Script) -> dict:
|
||||
return {
|
||||
"id": script.id,
|
||||
"name": script.name,
|
||||
"category": script.category,
|
||||
"content": script.content,
|
||||
"description": script.description,
|
||||
"created_by": script.created_by,
|
||||
"created_at": str(script.created_at) if script.created_at else None,
|
||||
"updated_at": str(script.updated_at) if script.updated_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _execution_to_dict(execution) -> dict:
|
||||
return {
|
||||
"id": execution.id,
|
||||
"script_id": execution.script_id,
|
||||
"command": execution.command,
|
||||
"server_ids": execution.server_ids,
|
||||
"status": execution.status,
|
||||
"results": execution.results,
|
||||
"operator": execution.operator,
|
||||
"started_at": str(execution.started_at) if execution.started_at else None,
|
||||
"completed_at": str(execution.completed_at) if execution.completed_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _credential_to_dict(credential: DbCredential) -> dict:
|
||||
"""Convert credential to dict (never expose password in API response)"""
|
||||
return {
|
||||
"id": credential.id,
|
||||
"name": credential.name,
|
||||
"db_type": credential.db_type,
|
||||
"host": credential.host,
|
||||
"port": credential.port,
|
||||
"username": credential.username,
|
||||
"database": credential.database,
|
||||
"created_at": str(credential.created_at) if credential.created_at else None,
|
||||
}
|
||||
@@ -0,0 +1,181 @@
|
||||
"""Nexus — Servers API Routes (CRUD + heartbeat + health check + push)
|
||||
Presentation layer — receives HTTP requests, delegates to ServerService/SyncService.
|
||||
"""
|
||||
|
||||
from typing import Optional, List
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
|
||||
from server.api.dependencies import get_server_service, get_sync_service, get_db
|
||||
from server.application.services.server_service import ServerService
|
||||
from server.application.services.sync_service import SyncService
|
||||
from server.domain.models import Server
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
router = APIRouter(prefix="/api/servers", tags=["servers"])
|
||||
|
||||
|
||||
# ── CRUD ──
|
||||
|
||||
@router.get("/", response_model=list)
|
||||
async def list_servers(
|
||||
category: Optional[str] = Query(None, description="Filter by category"),
|
||||
service: ServerService = Depends(get_server_service),
|
||||
):
|
||||
"""List all servers, optionally filtered by category"""
|
||||
servers = await service.list_servers(category)
|
||||
return [_server_to_dict(s) for s in servers]
|
||||
|
||||
|
||||
@router.get("/{id}", response_model=dict)
|
||||
async def get_server(
|
||||
id: int,
|
||||
service: ServerService = Depends(get_server_service),
|
||||
):
|
||||
"""Get a single server by ID"""
|
||||
server = await service.get_server(id)
|
||||
if not server:
|
||||
raise HTTPException(status_code=404, detail="Server not found")
|
||||
return _server_to_dict(server)
|
||||
|
||||
|
||||
@router.post("/", response_model=dict, status_code=201)
|
||||
async def create_server(
|
||||
server_data: dict,
|
||||
service: ServerService = Depends(get_server_service),
|
||||
):
|
||||
"""Create a new server"""
|
||||
server = Server(**server_data)
|
||||
created = await service.create_server(server)
|
||||
return _server_to_dict(created)
|
||||
|
||||
|
||||
@router.put("/{id}", response_model=dict)
|
||||
async def update_server(
|
||||
id: int,
|
||||
server_data: dict,
|
||||
service: ServerService = Depends(get_server_service),
|
||||
):
|
||||
"""Update an existing server"""
|
||||
server = await service.get_server(id)
|
||||
if not server:
|
||||
raise HTTPException(status_code=404, detail="Server not found")
|
||||
for key, value in server_data.items():
|
||||
if hasattr(server, key) and key != "id":
|
||||
setattr(server, key, value)
|
||||
updated = await service.update_server(server)
|
||||
return _server_to_dict(updated)
|
||||
|
||||
|
||||
@router.delete("/{id}", status_code=204)
|
||||
async def delete_server(
|
||||
id: int,
|
||||
service: ServerService = Depends(get_server_service),
|
||||
):
|
||||
"""Delete a server"""
|
||||
result = await service.delete_server(id)
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="Server not found")
|
||||
|
||||
|
||||
# ── Health Check ──
|
||||
|
||||
@router.post("/check", response_model=dict)
|
||||
async def check_servers(
|
||||
server_ids: List[int],
|
||||
service: ServerService = Depends(get_server_service),
|
||||
):
|
||||
"""Check health status of specified servers (Agent-based, no SSH)"""
|
||||
return await service.check_all_servers(server_ids)
|
||||
|
||||
|
||||
# ── Push ──
|
||||
|
||||
@router.post("/push", response_model=dict)
|
||||
async def push_to_servers(
|
||||
payload: dict,
|
||||
sync_service: SyncService = Depends(get_sync_service),
|
||||
):
|
||||
"""Push files to multiple servers (batch mode with concurrency control)
|
||||
|
||||
Body: {
|
||||
"server_ids": [1, 2, 3, ...],
|
||||
"source_path": "/path/to/files",
|
||||
"sync_mode": "incremental", // incremental|full|overwrite|checksum
|
||||
"operator": "admin",
|
||||
"batch_size": 50,
|
||||
"concurrency": 10
|
||||
}
|
||||
"""
|
||||
server_ids = payload.get("server_ids", [])
|
||||
source_path = payload.get("source_path")
|
||||
if not server_ids or not source_path:
|
||||
raise HTTPException(status_code=400, detail="server_ids and source_path required")
|
||||
|
||||
results = await sync_service.batch_push(
|
||||
server_ids=server_ids,
|
||||
source_path=source_path,
|
||||
sync_mode=payload.get("sync_mode", "incremental"),
|
||||
operator=payload.get("operator", "admin"),
|
||||
batch_size=payload.get("batch_size", 50),
|
||||
concurrency=payload.get("concurrency", 10),
|
||||
)
|
||||
return {sid: _sync_log_to_dict(log) for sid, log in results.items()}
|
||||
|
||||
|
||||
# ── Sync Logs ──
|
||||
|
||||
@router.get("/{id}/logs", response_model=list)
|
||||
async def get_server_logs(
|
||||
id: int,
|
||||
limit: int = Query(50, ge=1, le=200),
|
||||
sync_service: SyncService = Depends(get_sync_service),
|
||||
):
|
||||
"""Get sync logs for a specific server"""
|
||||
logs = await sync_service.get_sync_logs(id, limit)
|
||||
return [_sync_log_to_dict(log) for log in logs]
|
||||
|
||||
|
||||
# ── Helper functions ──
|
||||
|
||||
def _server_to_dict(server: Server) -> dict:
|
||||
"""Convert Server model to API response dict (hide sensitive fields)"""
|
||||
return {
|
||||
"id": server.id,
|
||||
"name": server.name,
|
||||
"domain": server.domain,
|
||||
"port": server.port,
|
||||
"username": server.username,
|
||||
"auth_method": server.auth_method,
|
||||
"agent_port": server.agent_port,
|
||||
"description": server.description,
|
||||
"target_path": server.target_path,
|
||||
"category": server.category,
|
||||
"ssh_key_configured": server.ssh_key_configured,
|
||||
"is_online": server.is_online,
|
||||
"last_heartbeat": str(server.last_heartbeat) if server.last_heartbeat else None,
|
||||
"system_info": server.system_info,
|
||||
"agent_version": server.agent_version,
|
||||
"created_at": str(server.created_at) if server.created_at else None,
|
||||
"updated_at": str(server.updated_at) if server.updated_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _sync_log_to_dict(log) -> dict:
|
||||
"""Convert SyncLog model to API response dict"""
|
||||
return {
|
||||
"id": log.id,
|
||||
"server_id": log.server_id,
|
||||
"source_path": log.source_path,
|
||||
"target_path": log.target_path,
|
||||
"trigger_type": log.trigger_type,
|
||||
"operator": log.operator,
|
||||
"status": log.status,
|
||||
"sync_mode": log.sync_mode,
|
||||
"files_total": log.files_total,
|
||||
"files_transferred": log.files_transferred,
|
||||
"duration_seconds": log.duration_seconds,
|
||||
"error_message": log.error_message,
|
||||
"started_at": str(log.started_at) if log.started_at else None,
|
||||
"finished_at": str(log.finished_at) if log.finished_at else None,
|
||||
}
|
||||
@@ -0,0 +1,216 @@
|
||||
"""Nexus — Settings & Schedules API Routes
|
||||
Presentation layer — system settings + push schedules + password presets + audit logs.
|
||||
"""
|
||||
|
||||
from typing import Optional, List
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
||||
from server.api.dependencies import get_db
|
||||
from server.infrastructure.database.setting_repo import SettingRepositoryImpl
|
||||
from server.infrastructure.database.push_schedule_repo import PushScheduleRepositoryImpl, PushRetryJobRepositoryImpl
|
||||
from server.infrastructure.database.password_preset_repo import PasswordPresetRepositoryImpl
|
||||
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
|
||||
from server.domain.models import Setting, PushSchedule, PushRetryJob, PasswordPreset, AuditLog
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
router = APIRouter(prefix="/api/settings", tags=["settings"])
|
||||
|
||||
|
||||
# ── System Settings ──
|
||||
|
||||
@router.get("/", response_model=list)
|
||||
async def list_settings(db: AsyncSession = Depends(get_db)):
|
||||
"""List all system settings"""
|
||||
repo = SettingRepositoryImpl(db)
|
||||
settings = await repo.get_all()
|
||||
return [{"key": s.key, "value": s.value, "updated_at": str(s.updated_at)} for s in settings]
|
||||
|
||||
|
||||
@router.get("/{key}", response_model=dict)
|
||||
async def get_setting(key: str, db: AsyncSession = Depends(get_db)):
|
||||
"""Get a single setting by key"""
|
||||
repo = SettingRepositoryImpl(db)
|
||||
value = await repo.get(key)
|
||||
if value is None:
|
||||
raise HTTPException(status_code=404, detail="Setting not found")
|
||||
return {"key": key, "value": value}
|
||||
|
||||
|
||||
@router.put("/{key}", response_model=dict)
|
||||
async def set_setting(key: str, payload: dict, db: AsyncSession = Depends(get_db)):
|
||||
"""Set a system setting value"""
|
||||
repo = SettingRepositoryImpl(db)
|
||||
value = payload.get("value")
|
||||
if value is None:
|
||||
raise HTTPException(status_code=400, detail="value required")
|
||||
setting = await repo.set(key, str(value))
|
||||
return {"key": setting.key, "value": setting.value}
|
||||
|
||||
|
||||
# ── Push Schedules ──
|
||||
|
||||
schedule_router = APIRouter(prefix="/api/schedules", tags=["schedules"])
|
||||
|
||||
|
||||
@schedule_router.get("/", response_model=list)
|
||||
async def list_schedules(db: AsyncSession = Depends(get_db)):
|
||||
"""List all push schedules"""
|
||||
repo = PushScheduleRepositoryImpl(db)
|
||||
schedules = await repo.get_all()
|
||||
return [_schedule_to_dict(s) for s in schedules]
|
||||
|
||||
|
||||
@schedule_router.post("/", response_model=dict, status_code=201)
|
||||
async def create_schedule(payload: dict, db: AsyncSession = Depends(get_db)):
|
||||
"""Create a new push schedule
|
||||
|
||||
Body: {
|
||||
"name": "Daily web push",
|
||||
"source_path": "/www/wwwroot/site",
|
||||
"server_ids": "[1,2,3]",
|
||||
"cron_expr": "0 2 * * *",
|
||||
"enabled": true
|
||||
}
|
||||
"""
|
||||
repo = PushScheduleRepositoryImpl(db)
|
||||
schedule = PushSchedule(**payload)
|
||||
created = await repo.create(schedule)
|
||||
return _schedule_to_dict(created)
|
||||
|
||||
|
||||
@schedule_router.put("/{id}", response_model=dict)
|
||||
async def update_schedule(id: int, payload: dict, db: AsyncSession = Depends(get_db)):
|
||||
"""Update a push schedule"""
|
||||
repo = PushScheduleRepositoryImpl(db)
|
||||
schedule = await repo.get_by_id(id)
|
||||
if not schedule:
|
||||
raise HTTPException(status_code=404, detail="Schedule not found")
|
||||
for key, value in payload.items():
|
||||
if hasattr(schedule, key) and key != "id":
|
||||
setattr(schedule, key, value)
|
||||
updated = await repo.update(schedule)
|
||||
return _schedule_to_dict(updated)
|
||||
|
||||
|
||||
@schedule_router.delete("/{id}", status_code=204)
|
||||
async def delete_schedule(id: int, db: AsyncSession = Depends(get_db)):
|
||||
"""Delete a push schedule"""
|
||||
repo = PushScheduleRepositoryImpl(db)
|
||||
result = await repo.delete(id)
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="Schedule not found")
|
||||
|
||||
|
||||
# ── Password Presets ──
|
||||
|
||||
preset_router = APIRouter(prefix="/api/presets", tags=["presets"])
|
||||
|
||||
|
||||
@preset_router.get("/", response_model=list)
|
||||
async def list_presets(db: AsyncSession = Depends(get_db)):
|
||||
"""List all password presets"""
|
||||
repo = PasswordPresetRepositoryImpl(db)
|
||||
presets = await repo.get_all()
|
||||
return [{"id": p.id, "name": p.name, "created_at": str(p.created_at)} for p in presets]
|
||||
|
||||
|
||||
@preset_router.post("/", response_model=dict, status_code=201)
|
||||
async def create_preset(payload: dict, db: AsyncSession = Depends(get_db)):
|
||||
"""Create a password preset (password will be encrypted)
|
||||
|
||||
Body: {"name": "root-default", "encrypted_pw": "actual_password"}
|
||||
"""
|
||||
from server.infrastructure.database.crypto import encrypt_value
|
||||
repo = PasswordPresetRepositoryImpl(db)
|
||||
payload["encrypted_pw"] = encrypt_value(payload.get("encrypted_pw", ""))
|
||||
preset = PasswordPreset(**payload)
|
||||
created = await repo.create(preset)
|
||||
return {"id": created.id, "name": created.name}
|
||||
|
||||
|
||||
@preset_router.delete("/{id}", status_code=204)
|
||||
async def delete_preset(id: int, db: AsyncSession = Depends(get_db)):
|
||||
"""Delete a password preset"""
|
||||
repo = PasswordPresetRepositoryImpl(db)
|
||||
result = await repo.delete(id)
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="Preset not found")
|
||||
|
||||
|
||||
# ── Audit Logs ──
|
||||
|
||||
audit_router = APIRouter(prefix="/api/audit", tags=["audit"])
|
||||
|
||||
|
||||
@audit_router.get("/", response_model=list)
|
||||
async def list_audit_logs(
|
||||
action: Optional[str] = None,
|
||||
limit: int = 200,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""List recent audit logs, optionally filtered by action"""
|
||||
repo = AuditLogRepositoryImpl(db)
|
||||
if action:
|
||||
logs = await repo.get_by_action(action, limit)
|
||||
else:
|
||||
logs = await repo.get_recent(limit)
|
||||
return [_audit_to_dict(log) for log in logs]
|
||||
|
||||
|
||||
# ── Retry Jobs ──
|
||||
|
||||
retry_router = APIRouter(prefix="/api/retries", tags=["retries"])
|
||||
|
||||
|
||||
@retry_router.get("/", response_model=list)
|
||||
async def list_retry_jobs(limit: int = 100, db: AsyncSession = Depends(get_db)):
|
||||
"""List pending retry jobs"""
|
||||
repo = PushRetryJobRepositoryImpl(db)
|
||||
jobs = await repo.get_pending(limit)
|
||||
return [_retry_to_dict(j) for j in jobs]
|
||||
|
||||
|
||||
# ── Helper functions ──
|
||||
|
||||
def _schedule_to_dict(schedule: PushSchedule) -> dict:
|
||||
return {
|
||||
"id": schedule.id,
|
||||
"name": schedule.name,
|
||||
"source_path": schedule.source_path,
|
||||
"server_ids": schedule.server_ids,
|
||||
"cron_expr": schedule.cron_expr,
|
||||
"enabled": schedule.enabled,
|
||||
"last_run_at": str(schedule.last_run_at) if schedule.last_run_at else None,
|
||||
"created_at": str(schedule.created_at) if schedule.created_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _audit_to_dict(log: AuditLog) -> dict:
|
||||
return {
|
||||
"id": log.id,
|
||||
"admin_username": log.admin_username,
|
||||
"action": log.action,
|
||||
"target_type": log.target_type,
|
||||
"target_id": log.target_id,
|
||||
"detail": log.detail,
|
||||
"ip_address": log.ip_address,
|
||||
"created_at": str(log.created_at) if log.created_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _retry_to_dict(job: PushRetryJob) -> dict:
|
||||
return {
|
||||
"id": job.id,
|
||||
"server_id": job.server_id,
|
||||
"server_name": job.server_name,
|
||||
"operator": job.operator,
|
||||
"source_path": job.source_path,
|
||||
"target_path": job.target_path,
|
||||
"status": job.status,
|
||||
"retry_count": job.retry_count,
|
||||
"max_retries": job.max_retries,
|
||||
"next_retry_at": str(job.next_retry_at) if job.next_retry_at else None,
|
||||
"last_error": job.last_error,
|
||||
"created_at": str(job.created_at) if job.created_at else None,
|
||||
}
|
||||
@@ -1 +1,8 @@
|
||||
"""Nexus — Application Services (Business logic)"""
|
||||
"""Nexus — Application Services Package"""
|
||||
|
||||
from server.application.services.server_service import ServerService
|
||||
from server.application.services.script_service import ScriptService
|
||||
from server.application.services.auth_service import AuthService
|
||||
from server.application.services.sync_service import SyncService
|
||||
|
||||
__all__ = ["ServerService", "ScriptService", "AuthService", "SyncService"]
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Nexus — Auth Service (Login / TOTP / session / brute-force protection)
|
||||
Application layer — orchestrates Admin Repository + LoginAttempt + TOTP + session.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import hashlib
|
||||
import hmac
|
||||
import secrets
|
||||
import struct
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
from server.domain.models import Admin, LoginAttempt, AuditLog
|
||||
from server.domain.repositories import AdminRepository, LoginAttemptRepository, AuditLogRepository
|
||||
|
||||
logger = logging.getLogger("nexus.auth_service")
|
||||
|
||||
# Brute-force protection thresholds
|
||||
MAX_LOGIN_FAILURES = 5 # Lock out after 5 failures within 15 minutes
|
||||
LOCKOUT_MINUTES = 15
|
||||
|
||||
|
||||
class AuthService:
|
||||
"""Business logic for authentication, TOTP, and session management"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
admin_repo: AdminRepository,
|
||||
attempt_repo: LoginAttemptRepository,
|
||||
audit_repo: AuditLogRepository,
|
||||
):
|
||||
self.admin_repo = admin_repo
|
||||
self.attempt_repo = attempt_repo
|
||||
self.audit_repo = audit_repo
|
||||
|
||||
async def login(self, username: str, password: str, ip_address: str, totp_code: Optional[str] = None) -> dict:
|
||||
"""Authenticate admin user with password + optional TOTP
|
||||
|
||||
Returns:
|
||||
{"success": True, "token": "..."} or {"success": False, "reason": "..."}
|
||||
"""
|
||||
# Check brute-force lockout
|
||||
failures = await self.attempt_repo.count_recent_failures(username, ip_address, LOCKOUT_MINUTES)
|
||||
if failures >= MAX_LOGIN_FAILURES:
|
||||
await self._audit("login_locked", "admin", 0, f"Account locked: {username} from {ip_address}")
|
||||
return {"success": False, "reason": "account_locked", "message": "登录尝试过多，请15分钟后重试"}
|
||||
|
||||
# Find admin
|
||||
admin = await self.admin_repo.get_by_username(username)
|
||||
if not admin:
|
||||
await self._record_attempt(username, ip_address, False)
|
||||
return {"success": False, "reason": "invalid_credentials", "message": "用户名或密码错误"}
|
||||
|
||||
if not admin.is_active:
|
||||
await self._record_attempt(username, ip_address, False)
|
||||
return {"success": False, "reason": "account_disabled", "message": "账户已被禁用"}
|
||||
|
||||
# Verify password
|
||||
if not self._verify_password(password, admin.password_hash):
|
||||
await self._record_attempt(username, ip_address, False)
|
||||
return {"success": False, "reason": "invalid_credentials", "message": "用户名或密码错误"}
|
||||
|
||||
# Verify TOTP (if enabled)
|
||||
if admin.totp_enabled:
|
||||
if not totp_code:
|
||||
return {"success": False, "reason": "totp_required", "message": "需要TOTP验证码"}
|
||||
if not self._verify_totp(totp_code, admin.totp_secret):
|
||||
await self._record_attempt(username, ip_address, False)
|
||||
return {"success": False, "reason": "invalid_totp", "message": "TOTP验证码错误"}
|
||||
|
||||
# Success — generate session token
|
||||
token = secrets.token_urlsafe(32)
|
||||
await self._record_attempt(username, ip_address, True)
|
||||
|
||||
# Update last_login
|
||||
admin.last_login = __import__("datetime").datetime.utcnow()
|
||||
await self.admin_repo.update(admin)
|
||||
|
||||
await self._audit("login_success", "admin", admin.id, f"Login: {username} from {ip_address}")
|
||||
return {"success": True, "token": token, "admin_id": admin.id, "username": admin.username}
|
||||
|
||||
async def verify_session(self, token: str) -> Optional[Admin]:
|
||||
"""Verify session token — currently uses Redis cache (TODO: implement Redis session store)"""
|
||||
# TODO: Look up token in Redis session store → return Admin
|
||||
return None
|
||||
|
||||
async def setup_totp(self, admin_id: int) -> dict:
|
||||
"""Generate TOTP secret for an admin user (before enabling)"""
|
||||
import base64
|
||||
secret = base64.b32encode(secrets.token_bytes(20)).decode()
|
||||
admin = await self.admin_repo.get_by_id(admin_id)
|
||||
if not admin:
|
||||
return {"success": False, "reason": "admin_not_found"}
|
||||
|
||||
admin.totp_secret = secret
|
||||
# Don't enable yet — user must verify once first
|
||||
await self.admin_repo.update(admin)
|
||||
|
||||
# Generate provisioning URI for authenticator apps
|
||||
from server.config import settings
|
||||
issuer = settings.SYSTEM_NAME
|
||||
uri = f"otpauth://totp/{issuer}:{admin.username}?secret={secret}&issuer={issuer}"
|
||||
|
||||
await self._audit("setup_totp", "admin", admin_id, f"TOTP setup initiated for {admin.username}")
|
||||
return {"success": True, "secret": secret, "uri": uri}
|
||||
|
||||
async def enable_totp(self, admin_id: int, totp_code: str) -> dict:
|
||||
"""Enable TOTP after user verifies with a code"""
|
||||
admin = await self.admin_repo.get_by_id(admin_id)
|
||||
if not admin or not admin.totp_secret:
|
||||
return {"success": False, "reason": "no_secret"}
|
||||
|
||||
if not self._verify_totp(totp_code, admin.totp_secret):
|
||||
return {"success": False, "reason": "invalid_totp", "message": "验证码错误，TOTP未启用"}
|
||||
|
||||
admin.totp_enabled = True
|
||||
await self.admin_repo.update(admin)
|
||||
await self._audit("enable_totp", "admin", admin_id, f"TOTP enabled for {admin.username}")
|
||||
return {"success": True, "message": "TOTP已启用"}
|
||||
|
||||
async def disable_totp(self, admin_id: int) -> dict:
|
||||
"""Disable TOTP for an admin user"""
|
||||
admin = await self.admin_repo.get_by_id(admin_id)
|
||||
if not admin:
|
||||
return {"success": False, "reason": "admin_not_found"}
|
||||
|
||||
admin.totp_enabled = False
|
||||
admin.totp_secret = None
|
||||
await self.admin_repo.update(admin)
|
||||
await self._audit("disable_totp", "admin", admin_id, f"TOTP disabled for {admin.username}")
|
||||
return {"success": True, "message": "TOTP已禁用"}
|
||||
|
||||
# ── Private helpers ──
|
||||
|
||||
def _verify_password(self, password: str, password_hash: str) -> bool:
|
||||
"""Verify password against stored hash (bcrypt-compatible)"""
|
||||
try:
|
||||
import bcrypt
|
||||
return bcrypt.checkpw(password.encode(), password_hash.encode())
|
||||
except ImportError:
|
||||
# Fallback: SHA256 comparison (for initial setup before bcrypt installed)
|
||||
return hashlib.sha256(password.encode()).hexdigest() == password_hash
|
||||
|
||||
def _verify_totp(self, code: str, secret: str) -> bool:
|
||||
"""Verify TOTP code using RFC 6238 (HOTP with time counter)"""
|
||||
import base64
|
||||
key = base64.b32decode(secret, casefold=True)
|
||||
# Current time step (30-second intervals)
|
||||
counter = int(time.time()) // 30
|
||||
|
||||
# Check current and ±1 window (allows 30s clock drift)
|
||||
for offset in range(-1, 2):
|
||||
c = counter + offset
|
||||
# HOTP algorithm
|
||||
msg = struct.pack(">Q", c)
|
||||
h = hmac.new(key, msg, hashlib.sha1).digest()
|
||||
offset_byte = h[-1] & 0x0F
|
||||
code_int = struct.unpack(">I", h[offset_byte:offset_byte + 4])[0] & 0x7FFFFFFF
|
||||
expected = str(code_int % 1000000).zfill(6)
|
||||
if expected == code.strip():
|
||||
return True
|
||||
return False
|
||||
|
||||
async def _record_attempt(self, username: str, ip_address: str, success: bool):
|
||||
"""Record login attempt for brute-force tracking"""
|
||||
attempt = LoginAttempt(username=username, ip_address=ip_address, success=success)
|
||||
await self.attempt_repo.create(attempt)
|
||||
|
||||
async def _audit(self, action: str, target_type: str, target_id: int, detail: str):
|
||||
log = AuditLog(action=action, target_type=target_type, target_id=target_id, detail=detail, ip_address="")
|
||||
await self.audit_repo.create(log)
|
||||
@@ -0,0 +1,251 @@
|
||||
"""Nexus — Script Service (Business logic for script library + remote execution)
|
||||
Application layer — orchestrates Repository + Agent /api/exec + DB credential substitution.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from typing import Optional, List, Dict
|
||||
|
||||
import httpx
|
||||
|
||||
from server.domain.models import Script, ScriptExecution, DbCredential, AuditLog
|
||||
from server.domain.repositories import (
|
||||
ScriptRepository, ScriptExecutionRepository,
|
||||
DbCredentialRepository, ServerRepository, AuditLogRepository,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("nexus.script_service")
|
||||
|
||||
|
||||
class ScriptService:
|
||||
"""Business logic for script CRUD and remote command execution via Agent /api/exec"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
script_repo: ScriptRepository,
|
||||
execution_repo: ScriptExecutionRepository,
|
||||
credential_repo: DbCredentialRepository,
|
||||
server_repo: ServerRepository,
|
||||
audit_repo: AuditLogRepository,
|
||||
):
|
||||
self.script_repo = script_repo
|
||||
self.execution_repo = execution_repo
|
||||
self.credential_repo = credential_repo
|
||||
self.server_repo = server_repo
|
||||
self.audit_repo = audit_repo
|
||||
|
||||
# ── Script CRUD ──
|
||||
|
||||
async def list_scripts(self, category: Optional[str] = None) -> List[Script]:
|
||||
if category:
|
||||
return await self.script_repo.get_by_category(category)
|
||||
return await self.script_repo.get_all()
|
||||
|
||||
async def get_script(self, id: int) -> Optional[Script]:
|
||||
return await self.script_repo.get_by_id(id)
|
||||
|
||||
async def create_script(self, script: Script) -> Script:
|
||||
created = await self.script_repo.create(script)
|
||||
await self._audit("create_script", "script", created.id, f"Created script: {script.name}")
|
||||
return created
|
||||
|
||||
async def update_script(self, script: Script) -> Script:
|
||||
updated = await self.script_repo.update(script)
|
||||
await self._audit("update_script", "script", updated.id, f"Updated script: {script.name}")
|
||||
return updated
|
||||
|
||||
async def delete_script(self, id: int) -> bool:
|
||||
result = await self.script_repo.delete(id)
|
||||
if result:
|
||||
await self._audit("delete_script", "script", id, f"Deleted script ID: {id}")
|
||||
return result
|
||||
|
||||
# ── Command Execution (Agent /api/exec) ──
|
||||
|
||||
async def execute_command(
|
||||
self,
|
||||
command: str,
|
||||
server_ids: List[int],
|
||||
script_id: Optional[int] = None,
|
||||
credential_id: Optional[int] = None,
|
||||
timeout: int = 30,
|
||||
operator: str = "admin",
|
||||
) -> ScriptExecution:
|
||||
"""Execute a shell command on multiple servers via Agent /api/exec.
|
||||
|
||||
If credential_id is provided, $DB_* variables are substituted in the command
|
||||
before sending to each Agent. Variable substitution happens on Nexus side
|
||||
so the password never appears in the script text.
|
||||
"""
|
||||
# Resolve DB credential variables (if provided)
|
||||
resolved_command = await self._substitute_db_vars(command, credential_id)
|
||||
|
||||
# Create execution record
|
||||
execution = ScriptExecution(
|
||||
script_id=script_id,
|
||||
command=resolved_command,
|
||||
server_ids=json.dumps(server_ids),
|
||||
status="running",
|
||||
operator=operator,
|
||||
)
|
||||
execution = await self.execution_repo.create(execution)
|
||||
|
||||
# Dispatch to agents concurrently (max 10 at a time)
|
||||
results: Dict[str, dict] = {}
|
||||
semaphore = asyncio.Semaphore(10)
|
||||
|
||||
async def _exec_on_server(server_id: int):
|
||||
server = await self.server_repo.get_by_id(server_id)
|
||||
if not server or not server.is_online:
|
||||
results[str(server_id)] = {
|
||||
"status": "skipped",
|
||||
"stdout": "",
|
||||
"stderr": f"Server offline or not found (ID={server_id})",
|
||||
"exit_code": -1,
|
||||
}
|
||||
return
|
||||
async with semaphore:
|
||||
try:
|
||||
result = await self._call_agent_exec(
|
||||
server.domain, server.agent_port, server.agent_api_key,
|
||||
resolved_command, timeout,
|
||||
)
|
||||
results[str(server_id)] = result
|
||||
except Exception as e:
|
||||
results[str(server_id)] = {
|
||||
"status": "error",
|
||||
"stdout": "",
|
||||
"stderr": str(e),
|
||||
"exit_code": -1,
|
||||
}
|
||||
|
||||
tasks = [_exec_on_server(sid) for sid in server_ids]
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
# Determine overall status
|
||||
failed_count = sum(1 for r in results.values() if r.get("exit_code", -1) != 0)
|
||||
status = "completed" if failed_count == 0 else "failed" if failed_count == len(server_ids) else "partial"
|
||||
|
||||
# Update execution record
|
||||
execution = await self.execution_repo.update_status(
|
||||
execution.id, status, json.dumps(results)
|
||||
)
|
||||
|
||||
await self._audit(
|
||||
"execute_command", "script_execution", execution.id,
|
||||
f"Executed on {len(server_ids)} servers: {status} ({failed_count} failed)",
|
||||
)
|
||||
return execution
|
||||
|
||||
async def get_execution(self, id: int) -> Optional[ScriptExecution]:
|
||||
return await self.execution_repo.get_by_id(id)
|
||||
|
||||
# ── DB Credential Management ──
|
||||
|
||||
async def list_credentials(self) -> List[DbCredential]:
|
||||
return await self.credential_repo.get_all()
|
||||
|
||||
async def create_credential(self, credential: DbCredential) -> DbCredential:
|
||||
from server.infrastructure.database.crypto import encrypt_value
|
||||
credential.encrypted_password = encrypt_value(credential.encrypted_password)
|
||||
created = await self.credential_repo.create(credential)
|
||||
await self._audit("create_credential", "db_credential", created.id, f"Created credential: {credential.name}")
|
||||
return created
|
||||
|
||||
async def delete_credential(self, id: int) -> bool:
|
||||
result = await self.credential_repo.delete(id)
|
||||
if result:
|
||||
await self._audit("delete_credential", "db_credential", id, f"Deleted credential ID: {id}")
|
||||
return result
|
||||
|
||||
async def test_credential_connection(self, credential: DbCredential) -> dict:
|
||||
"""Test database connection using provided credentials (before saving)"""
|
||||
# This runs a simple connectivity test via Agent /api/exec
|
||||
# e.g., mysql -h $host -u $user -p'$pass' -e 'SELECT 1'
|
||||
from server.infrastructure.database.crypto import decrypt_value
|
||||
|
||||
password = credential.encrypted_password # Not yet encrypted if testing before save
|
||||
if credential.id:
|
||||
password = decrypt_value(credential.encrypted_password)
|
||||
|
||||
test_cmd = f"mysql -h {credential.host} -P {credential.port} -u {credential.username} -p'{password}' -e 'SELECT 1' 2>/dev/null && echo 'OK' || echo 'FAIL'"
|
||||
# Pick any online server to test from (or the Nexus server itself)
|
||||
return {"status": "test_command_ready", "command": test_cmd}
|
||||
|
||||
# ── Private helpers ──
|
||||
|
||||
async def _substitute_db_vars(self, command: str, credential_id: Optional[int]) -> str:
|
||||
"""Replace $DB_USER/$DB_PASS/$DB_HOST/$DB_PORT/$DB_NAME in command"""
|
||||
if not credential_id:
|
||||
return command
|
||||
|
||||
credential = await self.credential_repo.get_by_id(credential_id)
|
||||
if not credential:
|
||||
return command
|
||||
|
||||
from server.infrastructure.database.crypto import decrypt_value
|
||||
password = decrypt_value(credential.encrypted_password)
|
||||
|
||||
replacements = {
|
||||
"$DB_USER": credential.username,
|
||||
"$DB_PASS": password,
|
||||
"$DB_HOST": credential.host,
|
||||
"$DB_PORT": str(credential.port),
|
||||
"$DB_NAME": credential.database or "",
|
||||
}
|
||||
for var, value in replacements.items():
|
||||
command = command.replace(var, value)
|
||||
|
||||
return command
|
||||
|
||||
async def _call_agent_exec(
|
||||
self, host: str, port: int, api_key: str, command: str, timeout: int = 30,
|
||||
) -> dict:
|
||||
"""Call Agent /api/exec endpoint to run a command on a remote server"""
|
||||
url = f"http://{host}:{port}/api/exec"
|
||||
headers = {"X-API-Key": api_key, "Content-Type": "application/json"}
|
||||
payload = {"command": command, "timeout": timeout}
|
||||
|
||||
async with httpx.AsyncClient(timeout=timeout + 5) as client:
|
||||
try:
|
||||
resp = await client.post(url, json=payload, headers=headers)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
return {
|
||||
"status": data.get("status", "unknown"),
|
||||
"stdout": data.get("stdout", ""),
|
||||
"stderr": data.get("stderr", ""),
|
||||
"exit_code": data.get("exit_code", -1),
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"status": "error",
|
||||
"stdout": "",
|
||||
"stderr": f"Agent returned HTTP {resp.status_code}",
|
||||
"exit_code": -1,
|
||||
}
|
||||
except httpx.TimeoutException:
|
||||
return {
|
||||
"status": "timeout",
|
||||
"stdout": "",
|
||||
"stderr": f"Agent timeout after {timeout}s",
|
||||
"exit_code": -1,
|
||||
}
|
||||
except httpx.ConnectError:
|
||||
return {
|
||||
"status": "connection_error",
|
||||
"stdout": "",
|
||||
"stderr": f"Cannot connect to Agent at {host}:{port}",
|
||||
"exit_code": -1,
|
||||
}
|
||||
|
||||
async def _audit(self, action: str, target_type: str, target_id: int, detail: str):
|
||||
"""Write audit log entry"""
|
||||
log = AuditLog(
|
||||
action=action,
|
||||
target_type=target_type,
|
||||
target_id=target_id,
|
||||
detail=detail,
|
||||
)
|
||||
await self.audit_repo.create(log)
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Nexus — Server Service (Business logic for server management)
|
||||
Application layer — orchestrates Repository + SSH pool + Redis cache.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from typing import Optional, List, Dict
|
||||
|
||||
from server.domain.models import Server, SyncLog
|
||||
from server.domain.repositories import ServerRepository, SyncLogRepository
|
||||
from server.infrastructure.ssh.pool import SSHConfig, exec_command
|
||||
from server.infrastructure.redis.client import get_redis
|
||||
|
||||
logger = logging.getLogger("nexus.server_service")
|
||||
|
||||
|
||||
class ServerService:
|
||||
"""Business logic for server CRUD, health check, and push operations"""
|
||||
|
||||
def __init__(self, server_repo: ServerRepository, sync_log_repo: SyncLogRepository):
|
||||
self.server_repo = server_repo
|
||||
self.sync_log_repo = sync_log_repo
|
||||
|
||||
async def list_servers(self, category: Optional[str] = None) -> List[Server]:
|
||||
"""List all servers, optionally filtered by category"""
|
||||
if category:
|
||||
return await self.server_repo.get_by_category(category)
|
||||
return await self.server_repo.get_all()
|
||||
|
||||
async def get_server(self, id: int) -> Optional[Server]:
|
||||
return await self.server_repo.get_by_id(id)
|
||||
|
||||
async def create_server(self, server: Server) -> Server:
|
||||
return await self.server_repo.create(server)
|
||||
|
||||
async def update_server(self, server: Server) -> Server:
|
||||
return await self.server_repo.update(server)
|
||||
|
||||
async def delete_server(self, id: int) -> bool:
|
||||
return await self.server_repo.delete(id)
|
||||
|
||||
async def update_heartbeat(self, id: int, data: dict) -> None:
|
||||
"""Process Agent heartbeat data"""
|
||||
is_online = data.get("is_online", True)
|
||||
system_info = json.dumps(data.get("system_info", {}))
|
||||
agent_version = data.get("agent_version", "")
|
||||
await self.server_repo.update_heartbeat(id, is_online, system_info, agent_version)
|
||||
|
||||
async def check_all_servers(self, server_ids: List[int]) -> Dict[int, dict]:
|
||||
"""Check health of multiple servers via Agent API (no SSH)"""
|
||||
results = {}
|
||||
# TODO: Use httpx to call each Agent's /health endpoint
|
||||
for id in server_ids:
|
||||
server = await self.server_repo.get_by_id(id)
|
||||
if server and server.is_online:
|
||||
results[id] = {"status": "online", "system_info": json.loads(server.system_info or "{}")}
|
||||
elif server:
|
||||
results[id] = {"status": "offline"}
|
||||
else:
|
||||
results[id] = {"status": "not_found"}
|
||||
return results
|
||||
|
||||
async def push_to_servers(
|
||||
self,
|
||||
server_ids: List[int],
|
||||
source_path: str,
|
||||
sync_mode: str = "incremental",
|
||||
operator: str = "admin",
|
||||
) -> Dict[int, SyncLog]:
|
||||
"""Push files to multiple servers in batches"""
|
||||
results = {}
|
||||
servers = []
|
||||
for id in server_ids:
|
||||
server = await self.server_repo.get_by_id(id)
|
||||
if server:
|
||||
servers.append(server)
|
||||
|
||||
# TODO: Implement batch push with asyncio.Semaphore(10)
|
||||
# batch_size=50, interval=30s
|
||||
for server in servers:
|
||||
target_path = server.target_path or f"/www/wwwroot/"
|
||||
log = SyncLog(
|
||||
server_id=server.id,
|
||||
source_path=source_path,
|
||||
target_path=target_path,
|
||||
trigger_type="batch",
|
||||
operator=operator,
|
||||
status="pending",
|
||||
sync_mode=sync_mode,
|
||||
)
|
||||
log = await self.sync_log_repo.create(log)
|
||||
results[server.id] = log
|
||||
|
||||
return results
|
||||
@@ -0,0 +1,288 @@
|
||||
"""Nexus — Sync Service (Batch push engine with rsync + Agent coordination)
|
||||
Application layer — orchestrates Server Repository + SSH pool + Redis progress + WebSocket broadcast.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import tempfile
|
||||
import os
|
||||
from typing import Optional, List, Dict
|
||||
|
||||
from server.domain.models import Server, SyncLog, AuditLog
|
||||
from server.domain.repositories import (
|
||||
ServerRepository, SyncLogRepository, AuditLogRepository,
|
||||
PushRetryJobRepository,
|
||||
)
|
||||
from server.infrastructure.ssh.pool import SSHConfig, exec_command
|
||||
from server.infrastructure.redis.client import get_redis
|
||||
|
||||
logger = logging.getLogger("nexus.sync_service")
|
||||
|
||||
# Batch push configuration
|
||||
DEFAULT_BATCH_SIZE = 50 # Servers per batch
|
||||
DEFAULT_CONCURRENCY = 10 # Max concurrent SSH per batch
|
||||
DEFAULT_BATCH_INTERVAL = 30 # Seconds between batches
|
||||
DEFAULT_TIMEOUT = 300 # 5 minutes per server
|
||||
|
||||
|
||||
class SyncService:
|
||||
"""Business logic for batch file push with rsync and progress tracking"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
server_repo: ServerRepository,
|
||||
sync_log_repo: SyncLogRepository,
|
||||
audit_repo: AuditLogRepository,
|
||||
retry_repo: PushRetryJobRepository,
|
||||
):
|
||||
self.server_repo = server_repo
|
||||
self.sync_log_repo = sync_log_repo
|
||||
self.audit_repo = audit_repo
|
||||
self.retry_repo = retry_repo
|
||||
|
||||
async def batch_push(
|
||||
self,
|
||||
server_ids: List[int],
|
||||
source_path: str,
|
||||
sync_mode: str = "incremental",
|
||||
operator: str = "admin",
|
||||
batch_size: int = DEFAULT_BATCH_SIZE,
|
||||
concurrency: int = DEFAULT_CONCURRENCY,
|
||||
batch_interval: int = DEFAULT_BATCH_INTERVAL,
|
||||
) -> Dict[int, SyncLog]:
|
||||
"""Push files to multiple servers in batches with concurrency control.
|
||||
|
||||
Flow:
|
||||
1. Divide server_ids into batches (50-100 per batch)
|
||||
2. Each batch: up to 10 concurrent rsync connections
|
||||
3. Batch interval: 30 seconds between batches
|
||||
4. Failed servers: add to retry queue (max 100 retries)
|
||||
5. Progress tracked in Redis for WebSocket broadcast
|
||||
"""
|
||||
results: Dict[int, SyncLog] = {}
|
||||
servers = []
|
||||
|
||||
# Resolve all servers
|
||||
for sid in server_ids:
|
||||
server = await self.server_repo.get_by_id(sid)
|
||||
if server:
|
||||
servers.append(server)
|
||||
else:
|
||||
logger.warning(f"Server ID {sid} not found, skipping")
|
||||
|
||||
# Divide into batches
|
||||
batches = [servers[i:i + batch_size] for i in range(0, len(servers), batch_size)]
|
||||
total = len(servers)
|
||||
completed = 0
|
||||
|
||||
redis = await get_redis()
|
||||
|
||||
# Initialize progress in Redis
|
||||
if redis:
|
||||
progress_key = f"sync:progress:{operator}:{int(asyncio.get_event_loop().time())}"
|
||||
await redis.set(progress_key, json.dumps({
|
||||
"total": total,
|
||||
"completed": 0,
|
||||
"failed": 0,
|
||||
"current_batch": 0,
|
||||
"total_batches": len(batches),
|
||||
"status": "running",
|
||||
}))
|
||||
|
||||
for batch_idx, batch in enumerate(batches):
|
||||
logger.info(f"Batch {batch_idx + 1}/{len(batches)}: pushing to {len(batch)} servers")
|
||||
|
||||
# Update progress
|
||||
if redis:
|
||||
await redis.set(progress_key, json.dumps({
|
||||
"total": total,
|
||||
"completed": completed,
|
||||
"failed": sum(1 for r in results.values() if r.status == "failed"),
|
||||
"current_batch": batch_idx + 1,
|
||||
"total_batches": len(batches),
|
||||
"status": "running",
|
||||
}))
|
||||
|
||||
# Execute batch with concurrency control
|
||||
semaphore = asyncio.Semaphore(concurrency)
|
||||
batch_tasks = []
|
||||
|
||||
async def _push_to_server(server: Server):
|
||||
async with semaphore:
|
||||
log = await self._push_single(server, source_path, sync_mode, operator)
|
||||
results[server.id] = log
|
||||
if log.status == "success":
|
||||
completed += 1
|
||||
elif log.status == "failed":
|
||||
# Add to retry queue
|
||||
await self._add_retry(server, source_path, operator)
|
||||
|
||||
for server in batch:
|
||||
batch_tasks.append(_push_to_server(server))
|
||||
|
||||
await asyncio.gather(*batch_tasks)
|
||||
|
||||
# Batch interval (pause between batches)
|
||||
if batch_idx < len(batches) - 1 and batch_interval > 0:
|
||||
logger.info(f"Batch {batch_idx + 1} complete, waiting {batch_interval}s before next batch")
|
||||
await asyncio.sleep(batch_interval)
|
||||
|
||||
# Final progress update
|
||||
if redis:
|
||||
await redis.set(progress_key, json.dumps({
|
||||
"total": total,
|
||||
"completed": completed,
|
||||
"failed": sum(1 for r in results.values() if r.status == "failed"),
|
||||
"current_batch": len(batches),
|
||||
"total_batches": len(batches),
|
||||
"status": "completed",
|
||||
}))
|
||||
|
||||
await self._audit(
|
||||
"batch_push", "sync", 0,
|
||||
f"Pushed {source_path} to {total} servers: {completed} success, {total - completed} failed",
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
async def _push_single(
|
||||
self, server: Server, source_path: str, sync_mode: str, operator: str,
|
||||
) -> SyncLog:
|
||||
"""Push files to a single server via rsync-over-SSH"""
|
||||
target_path = server.target_path or "/www/wwwroot/"
|
||||
|
||||
# Build rsync flags based on sync_mode
|
||||
rsync_flags = self._rsync_flags(sync_mode)
|
||||
|
||||
# Create sync log
|
||||
log = SyncLog(
|
||||
server_id=server.id,
|
||||
source_path=source_path,
|
||||
target_path=target_path,
|
||||
trigger_type="batch",
|
||||
operator=operator,
|
||||
status="running",
|
||||
sync_mode=sync_mode,
|
||||
)
|
||||
log = await self.sync_log_repo.create(log)
|
||||
|
||||
start_time = __import__("datetime").datetime.utcnow()
|
||||
|
||||
try:
|
||||
# Build SSH config from server model
|
||||
ssh_config = SSHConfig.from_server(server)
|
||||
|
||||
# Build rsync command
|
||||
# For SSH Key: use in-memory key written to temp file
|
||||
key_file = None
|
||||
ssh_opts = "-o StrictHostKeyChecking=no"
|
||||
|
||||
if server.ssh_key_configured and server.ssh_key_private:
|
||||
# Write decrypted key to temp file (600 permission)
|
||||
from server.infrastructure.database.crypto import decrypt_value
|
||||
key_content = decrypt_value(server.ssh_key_private)
|
||||
key_file = tempfile.NamedTemporaryFile(mode='w', suffix='_key', delete=False)
|
||||
key_file.write(key_content)
|
||||
key_file.close()
|
||||
os.chmod(key_file.name, 0o600)
|
||||
ssh_opts += f" -i {key_file.name}"
|
||||
elif server.auth_method == "password" and server.password:
|
||||
# Use sshpass with environment variable (not -p flag)
|
||||
# sshpass -e SSHPASS=<password> rsync -e "sshpass -e ssh ..."
|
||||
from server.infrastructure.database.crypto import decrypt_value
|
||||
password = decrypt_value(server.password)
|
||||
ssh_opts = f"sshpass -e ssh {ssh_opts}"
|
||||
|
||||
rsync_cmd = f"rsync {rsync_flags} -e 'ssh -p {server.port} {ssh_opts}' {source_path} {server.username}@{server.domain}:{target_path}"
|
||||
|
||||
# Execute rsync via SSH
|
||||
result = await exec_command(ssh_config, rsync_cmd, timeout=DEFAULT_TIMEOUT)
|
||||
|
||||
# Clean up temp key file
|
||||
if key_file:
|
||||
try:
|
||||
os.unlink(key_file.name)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
duration = int((__import__("datetime").datetime.utcnow() - start_time).total_seconds())
|
||||
|
||||
# Parse rsync output for file statistics
|
||||
stdout = result.get("stdout", "")
|
||||
files_transferred = self._parse_rsync_transferred(stdout)
|
||||
|
||||
if result.get("exit_code", -1) == 0:
|
||||
log = await self.sync_log_repo.update_status(
|
||||
log.id, "success",
|
||||
files_transferred=files_transferred,
|
||||
duration_seconds=duration,
|
||||
diff_summary=stdout[:2000],
|
||||
finished_at=__import__("datetime").datetime.utcnow(),
|
||||
)
|
||||
else:
|
||||
log = await self.sync_log_repo.update_status(
|
||||
log.id, "failed",
|
||||
error_message=result.get("stderr", "Unknown error")[:2000],
|
||||
duration_seconds=duration,
|
||||
finished_at=__import__("datetime").datetime.utcnow(),
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Push to server {server.id} ({server.name}) failed: {e}")
|
||||
log = await self.sync_log_repo.update_status(
|
||||
log.id, "failed",
|
||||
error_message=str(e)[:2000],
|
||||
finished_at=__import__("datetime").datetime.utcnow(),
|
||||
)
|
||||
|
||||
return log
|
||||
|
||||
async def get_sync_logs(self, server_id: int, limit: int = 50) -> List[SyncLog]:
|
||||
return await self.sync_log_repo.get_by_server_id(server_id, limit)
|
||||
|
||||
async def _add_retry(self, server: Server, source_path: str, operator: str):
|
||||
"""Add failed push to retry queue"""
|
||||
from server.domain.models import PushRetryJob
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
job = PushRetryJob(
|
||||
server_id=server.id,
|
||||
server_name=server.name,
|
||||
operator=operator,
|
||||
source_path=source_path,
|
||||
target_path=server.target_path or "/www/wwwroot/",
|
||||
status="pending",
|
||||
next_retry_at=datetime.utcnow() + timedelta(minutes=5),
|
||||
)
|
||||
await self.retry_repo.create(job)
|
||||
|
||||
# ── Private helpers ──
|
||||
|
||||
def _rsync_flags(self, sync_mode: str) -> str:
|
||||
"""Build rsync flags based on sync mode"""
|
||||
base = "-avz"
|
||||
mode_flags = {
|
||||
"incremental": base, # Default: only sync modified files
|
||||
"full": f"{base} --delete", # Sync + delete extraneous files
|
||||
"overwrite": f"{base} --inplace", # Direct overwrite without temp files
|
||||
"checksum": f"{base} --checksum", # Checksum-based comparison
|
||||
}
|
||||
return mode_flags.get(sync_mode, base)
|
||||
|
||||
def _parse_rsync_transferred(self, stdout: str) -> int:
|
||||
"""Parse rsync output to count transferred files"""
|
||||
# rsync -v output contains lines like: "sent X bytes received Y bytes Z bytes/sec"
|
||||
# and a summary line with file counts
|
||||
for line in stdout.split("\n"):
|
||||
if "Number of regular files transferred:" in line:
|
||||
try:
|
||||
return int(line.split(":")[1].strip())
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
# Fallback: count non-summary lines
|
||||
return 0
|
||||
|
||||
async def _audit(self, action: str, target_type: str, target_id: int, detail: str):
|
||||
log = AuditLog(action=action, target_type=target_type, target_id=target_id, detail=detail)
|
||||
await self.audit_repo.create(log)
|
||||
@@ -3,8 +3,12 @@ Protocol-based interfaces for dependency inversion.
|
||||
Concrete implementations in infrastructure/database/.
|
||||
"""
|
||||
|
||||
from typing import Protocol, Optional, List
|
||||
from server.domain.models import Server, SyncLog, Script, ScriptExecution, DbCredential
|
||||
from typing import Protocol, Optional, List, Dict
|
||||
from server.domain.models import (
|
||||
Server, SyncLog, Script, ScriptExecution, DbCredential,
|
||||
Admin, LoginAttempt, Setting, PasswordPreset,
|
||||
PushSchedule, PushRetryJob, AuditLog,
|
||||
)
|
||||
|
||||
|
||||
class ServerRepository(Protocol):
|
||||
@@ -17,6 +21,7 @@ class ServerRepository(Protocol):
|
||||
async def create(self, server: Server) -> Server: ...
|
||||
async def update(self, server: Server) -> Server: ...
|
||||
async def delete(self, id: int) -> bool: ...
|
||||
async def update_heartbeat(self, id: int, is_online: bool, system_info: str, agent_version: str) -> None: ...
|
||||
|
||||
|
||||
class SyncLogRepository(Protocol):
|
||||
@@ -27,6 +32,66 @@ class SyncLogRepository(Protocol):
|
||||
async def update_status(self, id: int, status: str, **kwargs) -> SyncLog: ...
|
||||
|
||||
|
||||
class AdminRepository(Protocol):
|
||||
"""Admin user data access abstract interface"""
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[Admin]: ...
|
||||
async def get_by_username(self, username: str) -> Optional[Admin]: ...
|
||||
async def create(self, admin: Admin) -> Admin: ...
|
||||
async def update(self, admin: Admin) -> Admin: ...
|
||||
|
||||
|
||||
class LoginAttemptRepository(Protocol):
|
||||
"""Login attempt data access abstract interface"""
|
||||
|
||||
async def create(self, attempt: LoginAttempt) -> LoginAttempt: ...
|
||||
async def count_recent_failures(self, username: str, ip_address: str, minutes: int = 15) -> int: ...
|
||||
|
||||
|
||||
class SettingRepository(Protocol):
|
||||
"""System setting data access abstract interface"""
|
||||
|
||||
async def get(self, key: str) -> Optional[str]: ...
|
||||
async def set(self, key: str, value: str) -> Setting: ...
|
||||
async def get_all(self) -> List[Setting]: ...
|
||||
|
||||
|
||||
class PasswordPresetRepository(Protocol):
|
||||
"""Password preset data access abstract interface"""
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[PasswordPreset]: ...
|
||||
async def get_all(self) -> List[PasswordPreset]: ...
|
||||
async def create(self, preset: PasswordPreset) -> PasswordPreset: ...
|
||||
async def delete(self, id: int) -> bool: ...
|
||||
|
||||
|
||||
class PushScheduleRepository(Protocol):
|
||||
"""Push schedule data access abstract interface"""
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[PushSchedule]: ...
|
||||
async def get_all(self) -> List[PushSchedule]: ...
|
||||
async def get_enabled(self) -> List[PushSchedule]: ...
|
||||
async def create(self, schedule: PushSchedule) -> PushSchedule: ...
|
||||
async def update(self, schedule: PushSchedule) -> PushSchedule: ...
|
||||
async def delete(self, id: int) -> bool: ...
|
||||
|
||||
|
||||
class PushRetryJobRepository(Protocol):
|
||||
"""Push retry job data access abstract interface"""
|
||||
|
||||
async def get_pending(self, limit: int = 100) -> List[PushRetryJob]: ...
|
||||
async def create(self, job: PushRetryJob) -> PushRetryJob: ...
|
||||
async def update_status(self, id: int, status: str, **kwargs) -> PushRetryJob: ...
|
||||
|
||||
|
||||
class AuditLogRepository(Protocol):
|
||||
"""Audit log data access abstract interface"""
|
||||
|
||||
async def create(self, log: AuditLog) -> AuditLog: ...
|
||||
async def get_recent(self, limit: int = 200) -> List[AuditLog]: ...
|
||||
async def get_by_action(self, action: str, limit: int = 50) -> List[AuditLog]: ...
|
||||
|
||||
|
||||
class ScriptRepository(Protocol):
|
||||
"""Script library data access abstract interface"""
|
||||
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Nexus — Admin & LoginAttempt Repository (Async SQLAlchemy)"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select, func
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import Admin, LoginAttempt
|
||||
|
||||
|
||||
class AdminRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[Admin]:
|
||||
result = await self.session.execute(select(Admin).where(Admin.id == id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def get_by_username(self, username: str) -> Optional[Admin]:
|
||||
result = await self.session.execute(select(Admin).where(Admin.username == username))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def create(self, admin: Admin) -> Admin:
|
||||
self.session.add(admin)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(admin)
|
||||
return admin
|
||||
|
||||
async def update(self, admin: Admin) -> Admin:
|
||||
await self.session.commit()
|
||||
await self.session.refresh(admin)
|
||||
return admin
|
||||
|
||||
|
||||
class LoginAttemptRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def create(self, attempt: LoginAttempt) -> LoginAttempt:
|
||||
self.session.add(attempt)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(attempt)
|
||||
return attempt
|
||||
|
||||
async def count_recent_failures(self, username: str, ip_address: str, minutes: int = 15) -> int:
|
||||
from datetime import datetime, timedelta
|
||||
cutoff = datetime.utcnow() - timedelta(minutes=minutes)
|
||||
result = await self.session.execute(
|
||||
select(func.count(LoginAttempt.id))
|
||||
.where(LoginAttempt.username == username)
|
||||
.where(LoginAttempt.ip_address == ip_address)
|
||||
.where(LoginAttempt.success == False)
|
||||
.where(LoginAttempt.attempted_at >= cutoff)
|
||||
)
|
||||
return result.scalar_one() or 0
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Nexus — AuditLog Repository (Async SQLAlchemy)"""
|
||||
|
||||
from typing import List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import AuditLog
|
||||
|
||||
|
||||
class AuditLogRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def create(self, log: AuditLog) -> AuditLog:
|
||||
self.session.add(log)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(log)
|
||||
return log
|
||||
|
||||
async def get_recent(self, limit: int = 200) -> List[AuditLog]:
|
||||
result = await self.session.execute(
|
||||
select(AuditLog).order_by(AuditLog.created_at.desc()).limit(limit)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_by_action(self, action: str, limit: int = 50) -> List[AuditLog]:
|
||||
result = await self.session.execute(
|
||||
select(AuditLog)
|
||||
.where(AuditLog.action == action)
|
||||
.order_by(AuditLog.created_at.desc())
|
||||
.limit(limit)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
@@ -0,0 +1,34 @@
|
||||
"""Nexus — DbCredential Repository (Async SQLAlchemy)"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import DbCredential
|
||||
|
||||
|
||||
class DbCredentialRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[DbCredential]:
|
||||
result = await self.session.execute(select(DbCredential).where(DbCredential.id == id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def get_all(self) -> List[DbCredential]:
|
||||
result = await self.session.execute(select(DbCredential).order_by(DbCredential.id))
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def create(self, credential: DbCredential) -> DbCredential:
|
||||
self.session.add(credential)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(credential)
|
||||
return credential
|
||||
|
||||
async def delete(self, id: int) -> bool:
|
||||
credential = await self.get_by_id(id)
|
||||
if credential:
|
||||
await self.session.delete(credential)
|
||||
await self.session.commit()
|
||||
return True
|
||||
return False
|
||||
@@ -0,0 +1,34 @@
|
||||
"""Nexus — PasswordPreset Repository (Async SQLAlchemy)"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import PasswordPreset
|
||||
|
||||
|
||||
class PasswordPresetRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[PasswordPreset]:
|
||||
result = await self.session.execute(select(PasswordPreset).where(PasswordPreset.id == id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def get_all(self) -> List[PasswordPreset]:
|
||||
result = await self.session.execute(select(PasswordPreset).order_by(PasswordPreset.id))
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def create(self, preset: PasswordPreset) -> PasswordPreset:
|
||||
self.session.add(preset)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(preset)
|
||||
return preset
|
||||
|
||||
async def delete(self, id: int) -> bool:
|
||||
preset = await self.get_by_id(id)
|
||||
if preset:
|
||||
await self.session.delete(preset)
|
||||
await self.session.commit()
|
||||
return True
|
||||
return False
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Nexus — PushSchedule & PushRetryJob Repository (Async SQLAlchemy)"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import PushSchedule, PushRetryJob
|
||||
|
||||
|
||||
class PushScheduleRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[PushSchedule]:
|
||||
result = await self.session.execute(select(PushSchedule).where(PushSchedule.id == id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def get_all(self) -> List[PushSchedule]:
|
||||
result = await self.session.execute(select(PushSchedule).order_by(PushSchedule.id))
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_enabled(self) -> List[PushSchedule]:
|
||||
result = await self.session.execute(
|
||||
select(PushSchedule).where(PushSchedule.enabled == True).order_by(PushSchedule.id)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def create(self, schedule: PushSchedule) -> PushSchedule:
|
||||
self.session.add(schedule)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(schedule)
|
||||
return schedule
|
||||
|
||||
async def update(self, schedule: PushSchedule) -> PushSchedule:
|
||||
await self.session.commit()
|
||||
await self.session.refresh(schedule)
|
||||
return schedule
|
||||
|
||||
async def delete(self, id: int) -> bool:
|
||||
schedule = await self.get_by_id(id)
|
||||
if schedule:
|
||||
await self.session.delete(schedule)
|
||||
await self.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class PushRetryJobRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_pending(self, limit: int = 100) -> List[PushRetryJob]:
|
||||
result = await self.session.execute(
|
||||
select(PushRetryJob)
|
||||
.where(PushRetryJob.status == "pending")
|
||||
.order_by(PushRetryJob.created_at)
|
||||
.limit(limit)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def create(self, job: PushRetryJob) -> PushRetryJob:
|
||||
self.session.add(job)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(job)
|
||||
return job
|
||||
|
||||
async def update_status(self, id: int, status: str, **kwargs) -> PushRetryJob:
|
||||
job = await self.session.get(PushRetryJob, id)
|
||||
if job:
|
||||
job.status = status
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(job, key):
|
||||
setattr(job, key, value)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(job)
|
||||
return job
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Nexus — Script & ScriptExecution Repository (Async SQLAlchemy)"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import Script, ScriptExecution
|
||||
|
||||
|
||||
class ScriptRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[Script]:
|
||||
result = await self.session.execute(select(Script).where(Script.id == id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def get_all(self) -> List[Script]:
|
||||
result = await self.session.execute(select(Script).order_by(Script.id))
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_by_category(self, category: str) -> List[Script]:
|
||||
result = await self.session.execute(
|
||||
select(Script).where(Script.category == category).order_by(Script.id)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def create(self, script: Script) -> Script:
|
||||
self.session.add(script)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(script)
|
||||
return script
|
||||
|
||||
async def update(self, script: Script) -> Script:
|
||||
await self.session.commit()
|
||||
await self.session.refresh(script)
|
||||
return script
|
||||
|
||||
async def delete(self, id: int) -> bool:
|
||||
script = await self.get_by_id(id)
|
||||
if script:
|
||||
await self.session.delete(script)
|
||||
await self.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class ScriptExecutionRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[ScriptExecution]:
|
||||
result = await self.session.execute(
|
||||
select(ScriptExecution).where(ScriptExecution.id == id)
|
||||
)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def create(self, execution: ScriptExecution) -> ScriptExecution:
|
||||
self.session.add(execution)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(execution)
|
||||
return execution
|
||||
|
||||
async def update_status(self, id: int, status: str, results: str) -> ScriptExecution:
|
||||
execution = await self.session.get(ScriptExecution, id)
|
||||
if execution:
|
||||
execution.status = status
|
||||
execution.results = results
|
||||
await self.session.commit()
|
||||
await self.session.refresh(execution)
|
||||
return execution
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Nexus — Server Repository (Async SQLAlchemy implementation)
|
||||
Concrete implementation of ServerRepository protocol.
|
||||
"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select, update, delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import Server
|
||||
|
||||
|
||||
class ServerRepositoryImpl:
|
||||
"""Async SQLAlchemy implementation of Server data access"""
|
||||
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_id(self, id: int) -> Optional[Server]:
|
||||
result = await self.session.execute(select(Server).where(Server.id == id))
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
async def get_all(self) -> List[Server]:
|
||||
result = await self.session.execute(select(Server).order_by(Server.id))
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_by_category(self, category: str) -> List[Server]:
|
||||
result = await self.session.execute(
|
||||
select(Server).where(Server.category == category).order_by(Server.id)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_online(self) -> List[Server]:
|
||||
result = await self.session.execute(
|
||||
select(Server).where(Server.is_online == True).order_by(Server.id)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def create(self, server: Server) -> Server:
|
||||
self.session.add(server)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(server)
|
||||
return server
|
||||
|
||||
async def update(self, server: Server) -> Server:
|
||||
await self.session.commit()
|
||||
await self.session.refresh(server)
|
||||
return server
|
||||
|
||||
async def delete(self, id: int) -> bool:
|
||||
server = await self.get_by_id(id)
|
||||
if server:
|
||||
await self.session.delete(server)
|
||||
await self.session.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
async def update_heartbeat(self, id: int, is_online: bool, system_info: str, agent_version: str) -> None:
|
||||
"""Update server online status from Agent heartbeat"""
|
||||
await self.session.execute(
|
||||
update(Server)
|
||||
.where(Server.id == id)
|
||||
.values(
|
||||
is_online=is_online,
|
||||
last_heartbeat=datetime.datetime.utcnow(),
|
||||
system_info=system_info,
|
||||
agent_version=agent_version,
|
||||
)
|
||||
)
|
||||
await self.session.commit()
|
||||
|
||||
|
||||
import datetime
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Nexus — Setting Repository (Async SQLAlchemy)"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import Setting
|
||||
|
||||
|
||||
class SettingRepositoryImpl:
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get(self, key: str) -> Optional[str]:
|
||||
result = await self.session.execute(select(Setting).where(Setting.key == key))
|
||||
setting = result.scalar_one_or_none()
|
||||
return setting.value if setting else None
|
||||
|
||||
async def set(self, key: str, value: str) -> Setting:
|
||||
result = await self.session.execute(select(Setting).where(Setting.key == key))
|
||||
setting = result.scalar_one_or_none()
|
||||
if setting:
|
||||
setting.value = value
|
||||
else:
|
||||
setting = Setting(key=key, value=value)
|
||||
self.session.add(setting)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(setting)
|
||||
return setting
|
||||
|
||||
async def get_all(self) -> List[Setting]:
|
||||
result = await self.session.execute(select(Setting).order_by(Setting.key))
|
||||
return list(result.scalars().all())
|
||||
@@ -0,0 +1,40 @@
|
||||
"""Nexus — SyncLog Repository (Async SQLAlchemy implementation)"""
|
||||
|
||||
from typing import Optional, List
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from server.domain.models import SyncLog
|
||||
|
||||
|
||||
class SyncLogRepositoryImpl:
|
||||
"""Async SQLAlchemy implementation of SyncLog data access"""
|
||||
|
||||
def __init__(self, session: AsyncSession):
|
||||
self.session = session
|
||||
|
||||
async def get_by_server_id(self, server_id: int, limit: int = 50) -> List[SyncLog]:
|
||||
result = await self.session.execute(
|
||||
select(SyncLog)
|
||||
.where(SyncLog.server_id == server_id)
|
||||
.order_by(SyncLog.started_at.desc())
|
||||
.limit(limit)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def create(self, log: SyncLog) -> SyncLog:
|
||||
self.session.add(log)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(log)
|
||||
return log
|
||||
|
||||
async def update_status(self, id: int, status: str, **kwargs) -> SyncLog:
|
||||
log = await self.session.get(SyncLog, id)
|
||||
if log:
|
||||
log.status = status
|
||||
for key, value in kwargs.items():
|
||||
if hasattr(log, key):
|
||||
setattr(log, key, value)
|
||||
await self.session.commit()
|
||||
await self.session.refresh(log)
|
||||
return log
|
||||
+23
-6
@@ -11,6 +11,19 @@ from fastapi.middleware.cors import CORSMiddleware
|
||||
from server.config import settings
|
||||
from server.infrastructure.database.session import init_db
|
||||
|
||||
# API routes
|
||||
from server.api.servers import router as servers_router
|
||||
from server.api.auth import router as auth_router
|
||||
from server.api.agent import router as agent_router
|
||||
from server.api.scripts import router as scripts_router
|
||||
from server.api.settings import (
|
||||
router as settings_router,
|
||||
schedule_router,
|
||||
preset_router,
|
||||
audit_router,
|
||||
retry_router,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("nexus")
|
||||
|
||||
|
||||
@@ -49,9 +62,13 @@ app.add_middleware(
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# TODO: Register routers (servers, auth, agent, scripts, credentials)
|
||||
# from server.api.servers import router as servers_router
|
||||
# from server.api.auth import router as auth_router
|
||||
# from server.api.agent import router as agent_router
|
||||
# from server.api.scripts import router as scripts_router
|
||||
# from server.api.credentials import router as credentials_router
|
||||
# Register all routers
|
||||
app.include_router(servers_router)
|
||||
app.include_router(auth_router)
|
||||
app.include_router(agent_router)
|
||||
app.include_router(scripts_router)
|
||||
app.include_router(settings_router)
|
||||
app.include_router(schedule_router)
|
||||
app.include_router(preset_router)
|
||||
app.include_router(audit_router)
|
||||
app.include_router(retry_router)
|
||||
Reference in New Issue
Block a user