GustavoNogueira/A-Team-Security-Infra-Agent-Engineer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: organize project structure into backend/frontend/nginx folders

Browse Source
This commit is contained in:
nogueiraguh
2026-02-27 18:22:50 -03:00
parent aa561a87f3
commit faf54ec611
8 changed files with 22 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
backend/Dockerfile Normal file
View File

@@ -0,0 +1,23 @@
FROM python:3.12-slim
WORKDIR /app
# Install system deps
RUN apt-get update && apt-get install -y --no-install-recommends \
curl gcc libffi-dev && \
rm -rf /var/lib/apt/lists/*
# Install Python deps
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy app
COPY app.py .
COPY cis_runner.py .
# Data volume
RUN mkdir -p /data
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

655
backend/app.py Normal file
View File

@@ -0,0 +1,655 @@
"""
OCI CIS AI Agent - Backend API
FastAPI with JWT auth, TOTP MFA, RBAC (admin/user/viewer), OCI CLI management,
Vector DB config, CIS report execution, chat agent, downloads, audit log.
"""
import os, json, uuid, hashlib, hmac, time, base64, struct, secrets, subprocess
import shutil, asyncio, sqlite3, logging, socket, re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, List, Dict, Any
from contextlib import contextmanager
from fastapi import (
FastAPI, HTTPException, Depends, Request, UploadFile, File, Form,
Query, BackgroundTasks
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, FileResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import jwt as pyjwt
# ── Config ────────────────────────────────────────────────────────────────────
APP_SECRET = os.environ.get("APP_SECRET", secrets.token_hex(32))
JWT_ALG = "HS256"
JWT_EXP_H = int(os.environ.get("JWT_EXPIRY_HOURS", "12"))
DATA = Path(os.environ.get("DATA_DIR", "/data"))
DB_PATH = DATA / "agent.db"
OCI_DIR = DATA / "oci_configs"
REPORTS = DATA / "reports"
for d in [DATA, OCI_DIR, REPORTS]:
d.mkdir(parents=True, exist_ok=True)
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("agent")
app = FastAPI(title="OCI CIS AI Agent", version="1.0.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True,
allow_methods=["*"], allow_headers=["*"])
security = HTTPBearer()
# ── Database ──────────────────────────────────────────────────────────────────
@contextmanager
def db():
conn = sqlite3.connect(str(DB_PATH))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
finally:
conn.close()
def init_db():
with db() as c:
c.executescript("""
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY, username TEXT UNIQUE NOT NULL, email TEXT,
password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'viewer',
mfa_secret TEXT, mfa_enabled INTEGER DEFAULT 0,
is_active INTEGER DEFAULT 1,
created_at TEXT DEFAULT (datetime('now')),
last_login TEXT
);
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now')),
expires_at TEXT NOT NULL, is_active INTEGER DEFAULT 1
);
CREATE TABLE IF NOT EXISTS oci_configs (
id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
tenancy_name TEXT NOT NULL, tenancy_ocid TEXT NOT NULL,
user_ocid TEXT NOT NULL, fingerprint TEXT NOT NULL,
region TEXT NOT NULL, key_file_path TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS reports (
id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
tenancy_name TEXT NOT NULL, config_id TEXT,
status TEXT DEFAULT 'pending', report_data TEXT,
html_path TEXT, json_path TEXT,
created_at TEXT DEFAULT (datetime('now')),
completed_at TEXT, error_msg TEXT
);
CREATE TABLE IF NOT EXISTS vector_db_configs (
id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
db_type TEXT NOT NULL, host TEXT NOT NULL, port INTEGER NOT NULL,
database_name TEXT, collection_name TEXT,
username TEXT, password_enc TEXT, api_key_enc TEXT,
extra_config TEXT, is_active INTEGER DEFAULT 1,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS chat_messages (
id TEXT PRIMARY KEY, session_id TEXT NOT NULL,
user_id TEXT NOT NULL, role TEXT NOT NULL,
content TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS audit_log (
id TEXT PRIMARY KEY, user_id TEXT, username TEXT,
action TEXT NOT NULL, resource TEXT, details TEXT,
ip TEXT, created_at TEXT DEFAULT (datetime('now'))
);
""")
adm = c.execute("SELECT id FROM users WHERE username='admin'").fetchone()
if not adm:
c.execute(
"INSERT INTO users (id,username,email,password_hash,role) VALUES (?,?,?,?,?)",
(str(uuid.uuid4()), "admin", "admin@local", _hash_pw("admin123"), "admin")
)
log.info("Default admin created: admin / admin123")
# ── Crypto ────────────────────────────────────────────────────────────────────
def _hash_pw(pw: str) -> str:
salt = secrets.token_hex(16)
h = hashlib.pbkdf2_hmac("sha256", pw.encode(), salt.encode(), 100_000)
return f"{salt}:{h.hex()}"
def _verify_pw(pw: str, stored: str) -> bool:
salt, h = stored.split(":")
return hmac.compare_digest(
hashlib.pbkdf2_hmac("sha256", pw.encode(), salt.encode(), 100_000).hex(), h
)
def _totp_secret() -> str:
return base64.b32encode(secrets.token_bytes(20)).decode()
def _totp_verify(secret: str, code: str, window: int = 1) -> bool:
key = base64.b32decode(secret)
now = int(time.time()) // 30
for off in range(-window, window + 1):
msg = struct.pack(">Q", now + off)
h = hmac.new(key, msg, hashlib.sha1).digest()
o = h[-1] & 0x0F
c = str((struct.unpack(">I", h[o:o+4])[0] & 0x7FFFFFFF) % 1_000_000).zfill(6)
if hmac.compare_digest(c, code):
return True
return False
def _totp_uri(secret: str, user: str) -> str:
return f"otpauth://totp/OCI-CIS-Agent:{user}?secret={secret}&issuer=OCI-CIS-Agent"
def _make_token(uid: str, role: str, sid: str) -> str:
return pyjwt.encode(
{"sub": uid, "role": role, "sid": sid,
"exp": datetime.utcnow() + timedelta(hours=JWT_EXP_H),
"iat": datetime.utcnow()},
APP_SECRET, algorithm=JWT_ALG
)
def _enc(v: str) -> str:
return base64.b64encode(v.encode()).decode()
def _dec(v: str) -> str:
return base64.b64decode(v.encode()).decode()
# ── Auth deps ─────────────────────────────────────────────────────────────────
async def current_user(cred: HTTPAuthorizationCredentials = Depends(security)):
try:
p = pyjwt.decode(cred.credentials, APP_SECRET, algorithms=[JWT_ALG])
except pyjwt.ExpiredSignatureError:
raise HTTPException(401, "Token expirado")
except pyjwt.InvalidTokenError:
raise HTTPException(401, "Token inválido")
with db() as c:
u = c.execute("SELECT * FROM users WHERE id=? AND is_active=1", (p["sub"],)).fetchone()
s = c.execute("SELECT * FROM sessions WHERE id=? AND is_active=1", (p["sid"],)).fetchone()
if not u or not s:
raise HTTPException(401, "Sessão inválida")
return dict(u)
def require(*roles):
async def dep(u=Depends(current_user)):
if u["role"] not in roles:
raise HTTPException(403, f"Requer role: {', '.join(roles)}")
return u
return dep
def _audit(uid: str, uname: str, action: str, resource: str = None,
details: str = None, ip: str = None):
with db() as c:
c.execute(
"INSERT INTO audit_log (id,user_id,username,action,resource,details,ip) VALUES (?,?,?,?,?,?,?)",
(str(uuid.uuid4()), uid, uname, action, resource, details, ip)
)
# ── Models ────────────────────────────────────────────────────────────────────
class LoginReq(BaseModel):
username: str; password: str; totp_code: Optional[str] = None
class RegisterReq(BaseModel):
username: str; email: str; password: str; role: str = "viewer"
class TOTPVerify(BaseModel):
totp_code: str
class ChangePwReq(BaseModel):
current_password: str; new_password: str
class VectorDBReq(BaseModel):
db_type: str; host: str; port: int
database_name: Optional[str] = None; collection_name: Optional[str] = None
username: Optional[str] = None; password: Optional[str] = None
api_key: Optional[str] = None; extra_config: Optional[Dict] = None
class ChatMsg(BaseModel):
message: str; session_id: Optional[str] = None
class RunReportReq(BaseModel):
config_id: str; regions: Optional[List[str]] = None
class UserUpdateReq(BaseModel):
email: Optional[str] = None; role: Optional[str] = None
is_active: Optional[bool] = None
# ── Auth endpoints ────────────────────────────────────────────────────────────
@app.post("/api/auth/login")
async def login(req: LoginReq, request: Request):
with db() as c:
u = c.execute("SELECT * FROM users WHERE username=? AND is_active=1",
(req.username,)).fetchone()
if not u or not _verify_pw(req.password, u["password_hash"]):
raise HTTPException(401, "Credenciais inválidas")
u = dict(u)
if u["mfa_enabled"]:
if not req.totp_code:
return {"mfa_required": True, "message": "Código MFA necessário"}
if not _totp_verify(u["mfa_secret"], req.totp_code):
raise HTTPException(401, "Código MFA inválido")
sid = str(uuid.uuid4())
exp = (datetime.utcnow() + timedelta(hours=JWT_EXP_H)).isoformat()
with db() as c:
c.execute("INSERT INTO sessions (id,user_id,expires_at) VALUES (?,?,?)",
(sid, u["id"], exp))
c.execute("UPDATE users SET last_login=datetime('now') WHERE id=?", (u["id"],))
_audit(u["id"], u["username"], "login", ip=request.client.host if request.client else None)
return {
"token": _make_token(u["id"], u["role"], sid),
"user": {"id":u["id"],"username":u["username"],"email":u["email"],
"role":u["role"],"mfa_enabled":bool(u["mfa_enabled"])}
}
@app.post("/api/auth/logout")
async def logout_ep(u=Depends(current_user)):
with db() as c:
c.execute("UPDATE sessions SET is_active=0 WHERE user_id=?", (u["id"],))
return {"ok": True}
@app.post("/api/auth/register")
async def register(req: RegisterReq, adm=Depends(require("admin"))):
if req.role not in ("admin","user","viewer"):
raise HTTPException(400, "Role inválida")
uid = str(uuid.uuid4())
with db() as c:
if c.execute("SELECT 1 FROM users WHERE username=?", (req.username,)).fetchone():
raise HTTPException(400, "Usuário já existe")
c.execute("INSERT INTO users (id,username,email,password_hash,role) VALUES (?,?,?,?,?)",
(uid, req.username, req.email, _hash_pw(req.password), req.role))
_audit(adm["id"], adm["username"], "create_user", uid, f"user={req.username} role={req.role}")
return {"id": uid, "username": req.username, "role": req.role}
@app.post("/api/auth/change-password")
async def change_pw(req: ChangePwReq, u=Depends(current_user)):
if not _verify_pw(req.current_password, u["password_hash"]):
raise HTTPException(400, "Senha atual incorreta")
with db() as c:
c.execute("UPDATE users SET password_hash=? WHERE id=?",
(_hash_pw(req.new_password), u["id"]))
return {"ok": True}
# ── MFA ───────────────────────────────────────────────────────────────────────
@app.post("/api/mfa/setup")
async def mfa_setup(u=Depends(current_user)):
sec = _totp_secret()
with db() as c:
c.execute("UPDATE users SET mfa_secret=? WHERE id=?", (sec, u["id"]))
return {"secret": sec, "uri": _totp_uri(sec, u["username"])}
@app.post("/api/mfa/verify")
async def mfa_verify(req: TOTPVerify, u=Depends(current_user)):
if not u.get("mfa_secret"):
raise HTTPException(400, "Chame /api/mfa/setup primeiro")
if not _totp_verify(u["mfa_secret"], req.totp_code):
raise HTTPException(400, "Código inválido")
with db() as c:
c.execute("UPDATE users SET mfa_enabled=1 WHERE id=?", (u["id"],))
return {"ok": True, "message": "MFA ativado"}
@app.post("/api/mfa/disable/{user_id}")
async def mfa_disable(user_id: str, adm=Depends(require("admin"))):
with db() as c:
c.execute("UPDATE users SET mfa_enabled=0,mfa_secret=NULL WHERE id=?", (user_id,))
_audit(adm["id"], adm["username"], "disable_mfa", user_id)
return {"ok": True}
# ── Users ─────────────────────────────────────────────────────────────────────
@app.get("/api/users")
async def list_users(u=Depends(require("admin"))):
with db() as c:
rows = c.execute(
"SELECT id,username,email,role,mfa_enabled,is_active,created_at,last_login FROM users"
).fetchall()
return [dict(r) for r in rows]
@app.get("/api/users/me")
async def me(u=Depends(current_user)):
return {k: u[k] for k in ("id","username","email","role","mfa_enabled")}
@app.put("/api/users/{uid}")
async def update_user(uid: str, req: UserUpdateReq, adm=Depends(require("admin"))):
sets, vals = [], []
if req.email is not None: sets.append("email=?"); vals.append(req.email)
if req.role is not None:
if req.role not in ("admin","user","viewer"): raise HTTPException(400, "Role inválida")
sets.append("role=?"); vals.append(req.role)
if req.is_active is not None: sets.append("is_active=?"); vals.append(int(req.is_active))
if sets:
vals.append(uid)
with db() as c:
c.execute(f"UPDATE users SET {','.join(sets)} WHERE id=?", vals)
_audit(adm["id"], adm["username"], "update_user", uid)
return {"ok": True}
@app.delete("/api/users/{uid}")
async def del_user(uid: str, adm=Depends(require("admin"))):
if uid == adm["id"]: raise HTTPException(400, "Não pode desativar a si mesmo")
with db() as c:
c.execute("UPDATE users SET is_active=0 WHERE id=?", (uid,))
_audit(adm["id"], adm["username"], "deactivate_user", uid)
return {"ok": True}
# ── OCI Config ────────────────────────────────────────────────────────────────
@app.post("/api/oci/config")
async def save_oci(
tenancy_name: str = Form(...), tenancy_ocid: str = Form(...),
user_ocid: str = Form(...), fingerprint: str = Form(...),
region: str = Form(...),
private_key: UploadFile = File(...),
public_key: Optional[UploadFile] = File(None),
u = Depends(require("admin","user"))
):
cid = str(uuid.uuid4())
cdir = OCI_DIR / cid; cdir.mkdir(parents=True)
kp = cdir / "oci_api_key.pem"
kp.write_bytes(await private_key.read()); kp.chmod(0o600)
if public_key:
(cdir / "oci_api_key_public.pem").write_bytes(await public_key.read())
(cdir / "config").write_text(
f"[DEFAULT]\nuser={user_ocid}\nfingerprint={fingerprint}\n"
f"tenancy={tenancy_ocid}\nregion={region}\nkey_file={kp}\n"
)
with db() as c:
c.execute(
"INSERT INTO oci_configs (id,user_id,tenancy_name,tenancy_ocid,user_ocid,fingerprint,region,key_file_path) VALUES (?,?,?,?,?,?,?,?)",
(cid, u["id"], tenancy_name, tenancy_ocid, user_ocid, fingerprint, region, str(kp))
)
_audit(u["id"], u["username"], "save_oci_config", cid, f"tenancy={tenancy_name}")
return {"id": cid, "tenancy_name": tenancy_name, "region": region}
@app.get("/api/oci/configs")
async def list_oci(u=Depends(current_user)):
with db() as c:
if u["role"] == "admin":
rows = c.execute("SELECT id,user_id,tenancy_name,tenancy_ocid,region,created_at FROM oci_configs").fetchall()
else:
rows = c.execute("SELECT id,user_id,tenancy_name,tenancy_ocid,region,created_at FROM oci_configs WHERE user_id=?", (u["id"],)).fetchall()
return [dict(r) for r in rows]
@app.delete("/api/oci/configs/{cid}")
async def del_oci(cid: str, u=Depends(require("admin","user"))):
with db() as c:
cfg = c.execute("SELECT * FROM oci_configs WHERE id=?", (cid,)).fetchone()
if not cfg: raise HTTPException(404)
if u["role"] != "admin" and cfg["user_id"] != u["id"]:
raise HTTPException(403)
c.execute("DELETE FROM oci_configs WHERE id=?", (cid,))
d = OCI_DIR / cid
if d.exists(): shutil.rmtree(d)
return {"ok": True}
@app.post("/api/oci/test/{cid}")
async def test_oci(cid: str, u=Depends(require("admin","user"))):
cp = OCI_DIR / cid / "config"
if not cp.exists():
return {"status":"error","message":"Config não encontrada"}
try:
r = subprocess.run(
["oci","iam","region","list","--config-file",str(cp),"--output","json"],
capture_output=True, text=True, timeout=30
)
if r.returncode == 0:
return {"status":"success","message":"Conexão OK","data":json.loads(r.stdout)}
return {"status":"error","message":r.stderr[:500]}
except FileNotFoundError:
return {"status":"error","message":"OCI CLI não instalado"}
except subprocess.TimeoutExpired:
return {"status":"error","message":"Timeout na conexão"}
except Exception as e:
return {"status":"error","message":str(e)}
@app.post("/api/oci/install-cli")
async def install_cli(u=Depends(require("admin"))):
try:
r = subprocess.run(["pip","install","oci-cli","--break-system-packages"],
capture_output=True, text=True, timeout=300)
ok = r.returncode == 0
_audit(u["id"], u["username"], "install_oci_cli", details="success" if ok else r.stderr[:200])
return {"status":"success" if ok else "error",
"message":"OCI CLI instalado!" if ok else r.stderr[:500]}
except Exception as e:
return {"status":"error","message":str(e)}
# ── Reports ───────────────────────────────────────────────────────────────────
@app.post("/api/reports/run")
async def run_report(req: RunReportReq, bg: BackgroundTasks, u=Depends(require("admin","user"))):
with db() as c:
cfg = c.execute("SELECT * FROM oci_configs WHERE id=?", (req.config_id,)).fetchone()
if not cfg: raise HTTPException(404, "Config não encontrada")
rid = str(uuid.uuid4())
c.execute(
"INSERT INTO reports (id,user_id,tenancy_name,config_id,status) VALUES (?,?,?,?,?)",
(rid, u["id"], cfg["tenancy_name"], req.config_id, "running")
)
bg.add_task(_exec_report, rid, dict(cfg), req.regions)
_audit(u["id"], u["username"], "run_report", rid)
return {"report_id": rid, "status": "running"}
async def _exec_report(rid: str, cfg: dict, regions: Optional[List[str]]):
rdir = REPORTS / rid; rdir.mkdir(parents=True, exist_ok=True)
config_path = str(OCI_DIR / cfg["id"] / "config")
try:
cmd = ["python3", "/app/cis_runner.py", "--config", config_path,
"--output", str(rdir), "--tenancy-name", cfg["tenancy_name"]]
if regions:
cmd += ["--regions", ",".join(regions)]
proc = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
jp = rdir / "report.json"; hp = rdir / "report.html"
with db() as c:
if proc.returncode == 0 and jp.exists():
c.execute(
"UPDATE reports SET status='completed',report_data=?,html_path=?,json_path=?,completed_at=datetime('now') WHERE id=?",
(jp.read_text()[:500_000], str(hp) if hp.exists() else None, str(jp), rid)
)
else:
c.execute(
"UPDATE reports SET status='failed',error_msg=?,completed_at=datetime('now') WHERE id=?",
((stderr.decode() if stderr else "Unknown")[:2000], rid)
)
except Exception as e:
with db() as c:
c.execute(
"UPDATE reports SET status='failed',error_msg=?,completed_at=datetime('now') WHERE id=?",
(str(e)[:2000], rid)
)
@app.get("/api/reports")
async def list_reports(u=Depends(current_user)):
with db() as c:
q = "SELECT id,user_id,tenancy_name,status,created_at,completed_at,error_msg FROM reports"
rows = c.execute(q + " ORDER BY created_at DESC").fetchall() if u["role"]=="admin" \
else c.execute(q + " WHERE user_id=? ORDER BY created_at DESC", (u["id"],)).fetchall()
return [dict(r) for r in rows]
@app.get("/api/reports/{rid}")
async def get_report(rid: str, u=Depends(current_user)):
with db() as c:
r = c.execute("SELECT * FROM reports WHERE id=?", (rid,)).fetchone()
if not r: raise HTTPException(404)
if u["role"] != "admin" and r["user_id"] != u["id"]: raise HTTPException(403)
d = dict(r)
if d.get("report_data"):
try: d["report_data"] = json.loads(d["report_data"])
except: pass
return d
@app.get("/api/reports/{rid}/html")
async def report_html(rid: str):
with db() as c:
r = c.execute("SELECT html_path FROM reports WHERE id=?", (rid,)).fetchone()
if not r or not r["html_path"] or not Path(r["html_path"]).exists():
raise HTTPException(404)
return FileResponse(r["html_path"], media_type="text/html")
@app.get("/api/reports/{rid}/download")
async def report_dl(rid: str, fmt: str = Query("json"), u=Depends(current_user)):
with db() as c:
r = c.execute("SELECT * FROM reports WHERE id=?", (rid,)).fetchone()
if not r: raise HTTPException(404)
p = r["json_path"] if fmt == "json" else r["html_path"]
if not p or not Path(p).exists(): raise HTTPException(404)
return FileResponse(p, filename=f"cis_{r['tenancy_name']}_{rid[:8]}.{fmt}")
# ── Vector DB ─────────────────────────────────────────────────────────────────
@app.post("/api/vectordb/config")
async def save_vdb(req: VectorDBReq, u=Depends(require("admin","user"))):
vid = str(uuid.uuid4())
with db() as c:
c.execute(
"INSERT INTO vector_db_configs (id,user_id,db_type,host,port,database_name,collection_name,username,password_enc,api_key_enc,extra_config) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(vid, u["id"], req.db_type, req.host, req.port, req.database_name,
req.collection_name, req.username,
_enc(req.password) if req.password else None,
_enc(req.api_key) if req.api_key else None,
json.dumps(req.extra_config) if req.extra_config else None)
)
_audit(u["id"], u["username"], "save_vectordb", vid, req.db_type)
return {"id": vid, "db_type": req.db_type, "host": req.host}
@app.get("/api/vectordb/configs")
async def list_vdb(u=Depends(current_user)):
with db() as c:
rows = c.execute(
"SELECT id,db_type,host,port,database_name,collection_name,is_active,created_at FROM vector_db_configs WHERE user_id=? OR ?='admin'",
(u["id"], u["role"])
).fetchall()
return [dict(r) for r in rows]
@app.post("/api/vectordb/test/{vid}")
async def test_vdb(vid: str, u=Depends(require("admin","user"))):
with db() as c:
cfg = c.execute("SELECT * FROM vector_db_configs WHERE id=?", (vid,)).fetchone()
if not cfg: raise HTTPException(404)
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM); s.settimeout(5)
ok = s.connect_ex((cfg["host"], cfg["port"])) == 0; s.close()
return {"status":"success" if ok else "error",
"message":f"{'Conectado' if ok else 'Falha'} {cfg['host']}:{cfg['port']}"}
except Exception as e:
return {"status":"error","message":str(e)}
@app.delete("/api/vectordb/configs/{vid}")
async def del_vdb(vid: str, u=Depends(require("admin","user"))):
with db() as c:
c.execute("DELETE FROM vector_db_configs WHERE id=?", (vid,))
return {"ok": True}
# ── Chat Agent ────────────────────────────────────────────────────────────────
@app.post("/api/chat")
async def chat(msg: ChatMsg, u=Depends(current_user)):
sid = msg.session_id or str(uuid.uuid4())
with db() as c:
c.execute("INSERT INTO chat_messages (id,session_id,user_id,role,content) VALUES (?,?,?,?,?)",
(str(uuid.uuid4()), sid, u["id"], "user", msg.message))
resp = _agent_respond(msg.message, u)
with db() as c:
c.execute("INSERT INTO chat_messages (id,session_id,user_id,role,content) VALUES (?,?,?,?,?)",
(str(uuid.uuid4()), sid, u["id"], "assistant", resp))
return {"session_id": sid, "response": resp}
def _agent_respond(msg: str, user: dict) -> str:
m = msg.lower().strip()
if any(k in m for k in ["status","health","como está","saúde"]):
cli = "✅ Instalado" if shutil.which("oci") else "❌ Não instalado"
with db() as c:
nc = c.execute("SELECT COUNT(*) n FROM oci_configs").fetchone()["n"]
nr = c.execute("SELECT COUNT(*) n FROM reports").fetchone()["n"]
nu = c.execute("SELECT COUNT(*) n FROM users WHERE is_active=1").fetchone()["n"]
return (f"📊 **Status do Sistema**\n\n"
f"• OCI CLI: {cli}\n• Configs OCI: {nc}\n• Relatórios: {nr}\n"
f"• Usuários ativos: {nu}\n• Servidor: ✅ Online")
if any(k in m for k in ["listar config","list config","mostrar config"]):
with db() as c:
rows = c.execute("SELECT tenancy_name,region,created_at FROM oci_configs").fetchall()
if not rows: return "Nenhuma configuração OCI. Vá até **Config OCI** para adicionar."
lines = ["📋 **Configurações OCI:**\n"]
for r in rows:
lines.append(f"• **{r['tenancy_name']}** — `{r['region']}` ({r['created_at']})")
return "\n".join(lines)
if any(k in m for k in ["ajuda","help","comandos"]):
return ("🤖 **Comandos disponíveis:**\n\n"
"• `status` — Status do sistema\n"
"• `listar configs` — Configurações OCI\n"
"• `verificar cis` — Visão geral dos checks CIS 3.0\n"
"• `gerar report` — Instruções para executar relatório\n"
"• `ajuda` — Esta mensagem\n\n"
"Use as abas laterais para navegar entre as funcionalidades.")
if any(k in m for k in ["cis","benchmark","checks","verificar"]):
return ("🔒 **CIS OCI Foundations Benchmark 3.0**\n\n"
"54 controles em 8 domínios:\n"
"• IAM (1.1–1.17): 17 controles\n"
"• Networking (2.1–2.8): 8 controles\n"
"• Compute (3.1–3.3): 3 controles\n"
"• Logging & Monitoring (4.1–4.18): 18 controles\n"
"• Storage Object (5.1.1–5.1.3): 3 controles\n"
"• Storage Block (5.2.1–5.2.2): 2 controles\n"
"• Storage FSS (5.3.1): 1 controle\n"
"• Asset Management (6.1–6.2): 2 controles\n\n"
"Configure uma conexão OCI e execute um relatório na aba **Report**.")
if any(k in m for k in ["report","relatório","executar","rodar","gerar"]):
return ("Para gerar um relatório CIS:\n\n"
"1. Vá em **Config OCI** e adicione suas credenciais\n"
"2. Vá em **Report** e selecione a configuração\n"
"3. Clique em **Executar CIS Compliance**\n"
"4. Acompanhe o status em **Downloads**\n\n"
"O relatório HTML será gerado automaticamente.")
return ("Entendi sua mensagem. Sou o **AI Agent de Infraestrutura & Segurança OCI**.\n\n"
"Posso ajudar com:\n"
"🔍 Compliance CIS Benchmark 3.0\n"
"📊 Relatórios de segurança\n"
"🔧 Configuração OCI e Vector DB\n"
"📋 Análise de findings\n\n"
"Digite **ajuda** para ver os comandos.")
@app.delete("/api/chat/{sid}")
async def clear_chat(sid: str, u=Depends(current_user)):
with db() as c:
c.execute("DELETE FROM chat_messages WHERE session_id=? AND user_id=?", (sid, u["id"]))
return {"ok": True}
# ── Downloads ─────────────────────────────────────────────────────────────────
@app.get("/api/downloads")
async def list_dl(u=Depends(current_user)):
with db() as c:
q = "SELECT id,tenancy_name,status,created_at,completed_at FROM reports WHERE status='completed'"
rows = c.execute(q + " ORDER BY created_at DESC").fetchall() if u["role"]=="admin" \
else c.execute(q + " AND user_id=? ORDER BY created_at DESC", (u["id"],)).fetchall()
res = []
for r in rows:
d = dict(r); d["files"] = []
rd = REPORTS / d["id"]
if rd.exists():
d["files"] = [{"name":f.name,"size":f.stat().st_size} for f in rd.iterdir() if f.is_file()]
res.append(d)
return res
@app.get("/api/downloads/{rid}/{fname}")
async def dl_file(rid: str, fname: str, u=Depends(current_user)):
p = REPORTS / rid / fname
if not p.exists(): raise HTTPException(404)
return FileResponse(p, filename=fname)
# ── Audit ─────────────────────────────────────────────────────────────────────
@app.get("/api/audit-log")
async def audit_log(limit: int = Query(100, le=500), u=Depends(require("admin"))):
with db() as c:
rows = c.execute("SELECT * FROM audit_log ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall()
return [dict(r) for r in rows]
# ── Health ────────────────────────────────────────────────────────────────────
@app.get("/api/health")
async def health():
return {"status":"ok","ts":datetime.utcnow().isoformat()}
@app.on_event("startup")
async def startup():
init_db()
log.info("OCI CIS AI Agent API started")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

169
backend/cis_runner.py Normal file
View File

@@ -0,0 +1,169 @@
#!/usr/bin/env python3
"""
CIS Runner Stub — Generates placeholder reports.
Replace this with the actual MCP server integration later.
Usage: python3 cis_runner.py --config /path/to/oci/config --output /path/to/dir --tenancy-name mytenancy [--regions r1,r2]
"""
import argparse, json, datetime, sys
from pathlib import Path
CHECKS = {
"1.1": {"id":"IAM-1","section":"Identity and Access Management","title":"Ensure service level admins are created to manage resources of particular service"},
"1.2": {"id":"IAM-2","section":"Identity and Access Management","title":"Ensure permissions on all resources are given only to the tenancy administrator group"},
"1.3": {"id":"IAM-3","section":"Identity and Access Management","title":"Ensure IAM administrators cannot update tenancy Administrators group"},
"1.4": {"id":"IAM-4","section":"Identity and Access Management","title":"Ensure IAM password policy requires minimum length of 14 or greater"},
"1.5": {"id":"IAM-5","section":"Identity and Access Management","title":"Ensure IAM password policy expires passwords within 365 days"},
"1.6": {"id":"IAM-6","section":"Identity and Access Management","title":"Ensure IAM password policy prevents password reuse"},
"1.7": {"id":"IAM-7","section":"Identity and Access Management","title":"Ensure MFA is enabled for all users with a console password"},
"1.8": {"id":"IAM-8","section":"Identity and Access Management","title":"Ensure user API keys rotate within 90 days"},
"1.9": {"id":"IAM-9","section":"Identity and Access Management","title":"Ensure user customer secret keys rotate within 90 days"},
"1.10": {"id":"IAM-10","section":"Identity and Access Management","title":"Ensure user auth tokens rotate within 90 days"},
"1.11": {"id":"IAM-11","section":"Identity and Access Management","title":"Ensure user IAM Database Passwords rotate within 90 days"},
"1.12": {"id":"IAM-12","section":"Identity and Access Management","title":"Ensure API keys are not created for tenancy administrator users"},
"1.13": {"id":"IAM-13","section":"Identity and Access Management","title":"Ensure all OCI IAM user accounts have a valid and current email address"},
"1.14": {"id":"IAM-14","section":"Identity and Access Management","title":"Ensure Instance Principal authentication is used"},
"1.15": {"id":"IAM-15","section":"Identity and Access Management","title":"Ensure storage service-level admins cannot delete resources they manage"},
"1.16": {"id":"IAM-16","section":"Identity and Access Management","title":"Ensure OCI IAM credentials unused for 45 days or more are disabled"},
"1.17": {"id":"IAM-17","section":"Identity and Access Management","title":"Ensure there is only one active API Key per user"},
"2.1": {"id":"NTW-1","section":"Networking","title":"Ensure no security lists allow ingress from 0.0.0.0/0 to port 22"},
"2.2": {"id":"NTW-2","section":"Networking","title":"Ensure no security lists allow ingress from 0.0.0.0/0 to port 3389"},
"2.3": {"id":"NTW-3","section":"Networking","title":"Ensure no network security groups allow ingress from 0.0.0.0/0 to port 22"},
"2.4": {"id":"NTW-4","section":"Networking","title":"Ensure no network security groups allow ingress from 0.0.0.0/0 to port 3389"},
"2.5": {"id":"NTW-5","section":"Networking","title":"Ensure the default security list of every VCN restricts all traffic except ICMP"},
"2.6": {"id":"NTW-6","section":"Networking","title":"Ensure Oracle Integration Cloud (OIC) access is restricted"},
"2.7": {"id":"NTW-7","section":"Networking","title":"Ensure Oracle Analytics Cloud (OAC) access is restricted or deployed within a VCN"},
"2.8": {"id":"NTW-8","section":"Networking","title":"Ensure Oracle Autonomous Shared Database access is restricted or deployed within a VCN"},
"3.1": {"id":"COM-1","section":"Compute","title":"Ensure Compute Instance Legacy Metadata service endpoint is disabled"},
"3.2": {"id":"COM-2","section":"Compute","title":"Ensure Secure Boot is enabled on Compute Instance"},
"3.3": {"id":"COM-3","section":"Compute","title":"Ensure In-transit Encryption is enabled on Compute Instance"},
"4.1": {"id":"LAM-1","section":"Logging and Monitoring","title":"Ensure default tags are used on resources"},
"4.2": {"id":"LAM-2","section":"Logging and Monitoring","title":"Create at least one notification topic and subscription"},
"4.3": {"id":"LAM-3","section":"Logging and Monitoring","title":"Ensure a notification is configured for Identity Provider changes"},
"4.4": {"id":"LAM-4","section":"Logging and Monitoring","title":"Ensure a notification is configured for IdP group mapping changes"},
"4.5": {"id":"LAM-5","section":"Logging and Monitoring","title":"Ensure a notification is configured for IAM group changes"},
"4.6": {"id":"LAM-6","section":"Logging and Monitoring","title":"Ensure a notification is configured for IAM policy changes"},
"4.7": {"id":"LAM-7","section":"Logging and Monitoring","title":"Ensure a notification is configured for user changes"},
"4.8": {"id":"LAM-8","section":"Logging and Monitoring","title":"Ensure a notification is configured for VCN changes"},
"4.9": {"id":"LAM-9","section":"Logging and Monitoring","title":"Ensure a notification is configured for changes to route tables"},
"4.10": {"id":"LAM-10","section":"Logging and Monitoring","title":"Ensure a notification is configured for security list changes"},
"4.11": {"id":"LAM-11","section":"Logging and Monitoring","title":"Ensure a notification is configured for network security group changes"},
"4.12": {"id":"LAM-12","section":"Logging and Monitoring","title":"Ensure a notification is configured for changes to network gateways"},
"4.13": {"id":"LAM-13","section":"Logging and Monitoring","title":"Ensure VCN flow logging is enabled for all subnets"},
"4.14": {"id":"LAM-14","section":"Logging and Monitoring","title":"Ensure Cloud Guard is enabled in the root compartment"},
"4.15": {"id":"LAM-15","section":"Logging and Monitoring","title":"Ensure a notification is configured for Cloud Guard problems detected"},
"4.16": {"id":"LAM-16","section":"Logging and Monitoring","title":"Ensure customer created CMK is rotated at least annually"},
"4.17": {"id":"LAM-17","section":"Logging and Monitoring","title":"Ensure write level Object Storage logging is enabled for all buckets"},
"4.18": {"id":"LAM-18","section":"Logging and Monitoring","title":"Ensure a notification is configured for Local OCI User Authentication"},
"5.1.1":{"id":"STO-1-1","section":"Storage - Object Storage","title":"Ensure no Object Storage buckets are publicly visible"},
"5.1.2":{"id":"STO-1-2","section":"Storage - Object Storage","title":"Ensure Object Storage Buckets are encrypted with a CMK"},
"5.1.3":{"id":"STO-1-3","section":"Storage - Object Storage","title":"Ensure Versioning is Enabled for Object Storage Buckets"},
"5.2.1":{"id":"STO-2-1","section":"Storage - Block Volumes","title":"Ensure Block Volumes are encrypted with Customer Managed Keys"},
"5.2.2":{"id":"STO-2-2","section":"Storage - Block Volumes","title":"Ensure Boot Volumes are encrypted with Customer Managed Key"},
"5.3.1":{"id":"STO-3-1","section":"Storage - File Storage Service","title":"Ensure File Storage Systems are encrypted with Customer Managed Keys"},
"6.1": {"id":"AM-1","section":"Asset Management","title":"Create at least one compartment in your tenancy"},
"6.2": {"id":"AM-2","section":"Asset Management","title":"Ensure no resources are created in the root compartment"},
}
def gen_html(findings, tenancy, output):
now = datetime.datetime.now()
sections = {}
for cid, ck in findings.items():
sec = ck["section"]
sections.setdefault(sec, {"total":0,"passed":0,"failed":0})
sections[sec]["total"] += 1
if ck["status"] == "PASS": sections[sec]["passed"] += 1
elif ck["status"] == "FAIL": sections[sec]["failed"] += 1
tot = sum(s["total"] for s in sections.values())
pas = sum(s["passed"] for s in sections.values())
fai = sum(s["failed"] for s in sections.values())
sec_rows = "".join(f'<tr><td><strong>{s}</strong></td><td>{v["total"]}</td>'
f'<td style="color:#dc2626;font-weight:700">{v["failed"]}</td>'
f'<td style="color:#16a34a;font-weight:700">{v["passed"]}</td></tr>' for s,v in sorted(sections.items()))
det_rows = ""
for cid in sorted(findings.keys(), key=lambda x: [int(p) if p.isdigit() else p for p in x.replace('.',' ').split()]):
ck = findings[cid]; st = ck["status"]
cls = "pass" if st=="PASS" else "fail" if st=="FAIL" else "review"
det_rows += f'<tr><td>{ck["id"]}</td><td>{cid}</td><td>{ck["title"]}</td><td><span class="b b-{cls}">{st}</span></td><td>{ck["section"]}</td></tr>'
html = f"""<!DOCTYPE html><html lang="en"><head><meta charset="UTF-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>Cloud Security Assessment - {tenancy}</title><style>
*{{margin:0;padding:0;box-sizing:border-box}}body{{font-family:'Segoe UI',system-ui,sans-serif;background:#f8f9fa;color:#1a1a2e;line-height:1.6}}
.c{{max-width:1200px;margin:0 auto;padding:2rem}}h1{{font-size:2.5rem;color:#1a1a2e;margin-bottom:.5rem}}
h2{{color:#2d3748;margin:2rem 0 1rem;padding-bottom:.5rem;border-bottom:2px solid #e2e8f0}}
.m{{color:#718096;margin-bottom:2rem}}.d{{background:#fff;border:1px solid #e2e8f0;border-radius:8px;padding:1.5rem;margin:2rem 0}}
.d h3{{color:#c53030;margin-bottom:.5rem}}.d p{{font-size:.85rem;color:#4a5568}}
table{{width:100%;border-collapse:collapse;background:#fff;border-radius:8px;overflow:hidden;box-shadow:0 1px 3px rgba(0,0,0,.1);margin:1rem 0}}
th{{background:#1a365d;color:#fff;padding:.75rem 1rem;text-align:left;font-size:.8rem;text-transform:uppercase;letter-spacing:.05em}}
td{{padding:.75rem 1rem;border-bottom:1px solid #e2e8f0}}tr:hover{{background:#f7fafc}}
.b{{padding:.25rem .75rem;border-radius:12px;font-size:.8rem;font-weight:600}}
.b-pass{{background:#c6f6d5;color:#22543d}}.b-fail{{background:#fed7d7;color:#9b2c2c}}.b-review{{background:#fefcbf;color:#744210}}
.sc{{display:grid;grid-template-columns:repeat(3,1fr);gap:1rem;margin:1.5rem 0}}
.sd{{background:#fff;border-radius:8px;padding:1.5rem;box-shadow:0 1px 3px rgba(0,0,0,.1);text-align:center}}
.sd .n{{font-size:2.5rem;font-weight:700}}.sd .l{{color:#718096;font-size:.9rem}}
.ft{{text-align:center;color:#a0aec0;margin-top:3rem;padding:1rem;border-top:1px solid #e2e8f0}}
</style></head><body><div class="c">
<h1>Cloud Security Assessment</h1>
<div class="m"><strong>Tenancy:</strong> {tenancy}<br>Tenancy – Cloud Infrastructure<br><br>
{now.strftime('%B, %Y')}, Version [1.0]<br>Extract date: {now.isoformat()[:19]}</div>
<div class="d"><h3>Disclaimer</h3>
<p>This document, in any form, software or printed matter, contains proprietary information that is the exclusive property of Oracle.
Your access to and use of this confidential material is subject to the terms and conditions of your Oracle software license and service agreement,
which has been executed and with which you agree to comply.</p>
<p style="margin-top:.5rem">This document is for informational purposes only and is intended solely to assist you in planning for the implementation and
upgrade of the product features described. It is not a commitment to deliver any material, code, or functionality.</p></div>
<h2>Findings Overview</h2><p style="color:#718096">Resumo por domínio (Section)</p>
<div class="sc"><div class="sd"><div class="n" style="color:#2d3748">{tot}</div><div class="l">Total Controls</div></div>
<div class="sd"><div class="n" style="color:#16a34a">{pas}</div><div class="l">Passed</div></div>
<div class="sd"><div class="n" style="color:#dc2626">{fai}</div><div class="l">Failed</div></div></div>
<table><thead><tr><th>Domains</th><th>Total Controls</th><th>Failed</th><th>Passed</th></tr></thead><tbody>{sec_rows}</tbody></table>
<p style="color:#718096;margin-top:.5rem">Total: {tot} · Passed: {pas} · Failed: {fai}</p>
<h2>Detailed Findings</h2>
<table><thead><tr><th>ID</th><th>Check #</th><th>Title</th><th>Status</th><th>Section</th></tr></thead><tbody>{det_rows}</tbody></table>
<div class="ft">Generated by OCI CIS AI Agent · CIS OCI Foundations Benchmark 3.0 · {now.strftime('%Y-%m-%d %H:%M:%S')}</div>
</div></body></html>"""
Path(output).write_text(html, encoding="utf-8")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--config", required=True)
ap.add_argument("--output", required=True)
ap.add_argument("--tenancy-name", default="Tenancy")
ap.add_argument("--regions", default=None)
args = ap.parse_args()
out = Path(args.output); out.mkdir(parents=True, exist_ok=True)
# TODO: Replace with MCP server call
# For now generate a stub report showing all checks as REVIEW
print(f"[CIS Runner] Config: {args.config}")
print(f"[CIS Runner] Output: {args.output}")
print(f"[CIS Runner] Tenancy: {args.tenancy_name}")
# Check if OCI config file exists
if not Path(args.config).exists():
print(f"[ERROR] Config not found: {args.config}", file=sys.stderr)
sys.exit(1)
findings = {}
for cid, ck in CHECKS.items():
findings[cid] = {**ck, "status": "REVIEW", "findings": [], "total": []}
report = {
"tenancy": args.tenancy_name,
"generated_at": datetime.datetime.now().isoformat(),
"benchmark": "CIS OCI Foundations Benchmark 3.0",
"regions": args.regions.split(",") if args.regions else ["all"],
"summary": {
"total": len(findings),
"passed": sum(1 for f in findings.values() if f["status"]=="PASS"),
"failed": sum(1 for f in findings.values() if f["status"]=="FAIL"),
"review": sum(1 for f in findings.values() if f["status"]=="REVIEW"),
},
"findings": findings,
}
(out / "report.json").write_text(json.dumps(report, indent=2, default=str))
gen_html(findings, args.tenancy_name, str(out / "report.html"))
print("[CIS Runner] Reports generated successfully")
if __name__ == "__main__":
main()

5
backend/requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
fastapi==0.115.0
uvicorn[standard]==0.30.0
python-multipart==0.0.12
PyJWT==2.9.0
python-dotenv==1.0.1