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

feat: multimodal chat, region-specific MCP scanning, and resilience improvements

Browse Source 
Add image/PDF/text file upload in chat via new /api/chat/upload endpoint
with OCI GenAI ImageContent and DocumentContent support. Frontend shows
attach button, file preview, and handles multipart upload seamlessly.

Add optional regions parameter to all CIS MCP scan tools for targeting
specific OCI regions instead of full tenancy. Session cache scoped by
config+regions for concurrent multi-region scans.

Add orphaned report auto-detection on progress poll (marks as failed if
process not found). Increase nginx proxy timeout to 15 minutes. Improve
$api error handling for non-JSON responses (504 HTML pages).

Update README to v1.9.
This commit is contained in:
nogueiraguh
2026-03-05 13:27:17 -03:00
parent 1d52234b1a
commit ef43eaa7ba
5 changed files with 221 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
backend/mcp_cis_server.py
View File

@@ -25,10 +25,17 @@ from mcp.types import Tool, TextContent
# ──────────────────────────────────────────────────────────────────────
server = Server("cis-compliance")
# config_id -> {checker, sections_collected: set, sections_analyzed: set, ...}
# session_key -> {checker, sections_collected: set, sections_analyzed: set, ...}
_sessions: dict = {}
SESSION_TTL = 1800
def _session_key(config_id: str, regions: list[str] | None = None) -> str:
"""Build cache key that includes region filter so different scopes are cached separately."""
if not regions:
return config_id
return f"{config_id}:{','.join(sorted(regions))}"
OCI_CONFIGS_DIR = os.environ.get("OCI_CONFIGS_DIR", "/data/oci_configs")
# ──────────────────────────────────────────────────────────────────────
@@ -164,12 +171,13 @@ def _list_available_configs() -> list[dict]:
def _get_or_create_checker(config_id: str, regions: list[str] | None = None):
"""Return cached session or create a new CIS_Report instance (fast — no data collection)."""
key = _session_key(config_id, regions)
now = time.time()
if config_id in _sessions:
s = _sessions[config_id]
if key in _sessions:
s = _sessions[key]
if now - s["created_at"] < SESSION_TTL:
return s
del _sessions[config_id]
del _sessions[key]
import cis_reports
@@ -197,9 +205,10 @@ def _get_or_create_checker(config_id: str, regions: list[str] | None = None):
session = {
"checker": checker, "created_at": now, "config_id": config_id,
"regions": regions, "session_key": key,
"base_collected": False, "sections_collected": set(), "sections_analyzed": set(),
}
_sessions[config_id] = session
_sessions[key] = session
return session
@@ -446,37 +455,44 @@ TOOLS = [
description="Coleta dados IAM e executa checks CIS 1.1-1.17: políticas, usuários, MFA, API keys, passwords, dynamic groups. Rápido (~30s).",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões específicas (ex: ['us-ashburn-1']). Vazio = todas."},
}, "required": ["config_id"]}),
Tool(name="cis_scan_networking",
description="Coleta dados de rede e executa checks CIS 2.1-2.8: security lists, NSGs, VCNs, subnets, OIC, OAC, ADB.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões específicas (ex: ['us-ashburn-1']). Vazio = todas."},
}, "required": ["config_id"]}),
Tool(name="cis_scan_compute",
description="Coleta instâncias Compute e executa checks CIS 3.1-3.3: metadata v2, secure boot, in-transit encryption. Rápido.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões específicas (ex: ['us-ashburn-1']). Vazio = todas."},
}, "required": ["config_id"]}),
Tool(name="cis_scan_logging_monitoring",
description="Coleta dados de logging/monitoring e executa checks CIS 4.1-4.18: tags, notificações, events, Cloud Guard, KMS, logs.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões específicas (ex: ['us-ashburn-1']). Vazio = todas."},
}, "required": ["config_id"]}),
Tool(name="cis_scan_storage",
description="Coleta dados de storage e executa checks CIS 5.1.1-5.3.1: buckets, block/boot volumes, file storage, CMK, versioning.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões específicas (ex: ['us-ashburn-1']). Vazio = todas."},
}, "required": ["config_id"]}),
Tool(name="cis_scan_asset_management",
description="Coleta dados de assets e executa checks CIS 6.1-6.2: compartments, recursos no root.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões específicas (ex: ['us-ashburn-1']). Vazio = todas."},
}, "required": ["config_id"]}),
Tool(name="cis_get_check",
description="Retorna findings detalhados de um check CIS individual por ID (ex: '1.7', '2.1'). Requer scan da seção correspondente.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"check_id": {"type": "string", "description": "ID do check CIS (ex: '1.7')"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões específicas (ex: ['us-ashburn-1']). Vazio = todas."},
}, "required": ["config_id", "check_id"]}),
Tool(name="cis_get_remediation",
description="Orientação de remediação para um check CIS: descrição, passos, link docs Oracle. Estático, sem conexão OCI.",
@@ -484,14 +500,16 @@ TOOLS = [
"check_id": {"type": "string", "description": "ID do check CIS (ex: '1.7')"},
}, "required": ["check_id"]}),
Tool(name="cis_get_scan_status",
description="Status da sessão: quais seções já foram coletadas/analisadas, TTL restante.",
description="Status da sessão: quais seções já foram coletadas/analisadas, TTL restante, regiões filtradas.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões da sessão a consultar. Vazio = sessão sem filtro."},
}, "required": ["config_id"]}),
Tool(name="cis_invalidate_cache",
description="Limpa cache de sessão. Próximo scan coletará dados novamente.",
inputSchema={"type": "object", "properties": {
"config_id": {"type": "string", "description": "ID da config OCI"},
"regions": {"type": "array", "items": {"type": "string"}, "description": "Regiões da sessão a limpar. Vazio = sessão sem filtro."},
}, "required": ["config_id"]}),
]
@@ -515,16 +533,21 @@ def _handle_cis_list_checks(args):
def _handle_scan_section(args, section: str):
config_id = args["config_id"]
session = _get_or_create_checker(config_id)
regions = args.get("regions") or None
session = _get_or_create_checker(config_id, regions=regions)
_analyze_section(session, section)
return _section_summary(session["checker"], section)
result = _section_summary(session["checker"], section)
if regions:
result["regions_filter"] = regions
return result
def _handle_cis_get_check(args):
config_id = args["config_id"]; check_id = args["check_id"]
regions = args.get("regions") or None
section = _check_id_to_section(check_id)
if section == "unknown":
return {"error": f"Check '{check_id}' não encontrado. Use cis_list_checks."}
session = _get_or_create_checker(config_id)
session = _get_or_create_checker(config_id, regions=regions)
_analyze_section(session, section)
benchmark = _get_benchmark(session["checker"])
if check_id not in benchmark:
@@ -549,12 +572,14 @@ def _handle_cis_get_remediation(args):
def _handle_cis_get_scan_status(args):
config_id = args["config_id"]
if config_id not in _sessions:
return {"config_id": config_id, "status": "not_initialized", "sections_collected": [], "sections_analyzed": []}
s = _sessions[config_id]
regions = args.get("regions") or None
key = _session_key(config_id, regions)
if key not in _sessions:
return {"config_id": config_id, "regions": regions, "status": "not_initialized", "sections_collected": [], "sections_analyzed": []}
s = _sessions[key]
elapsed = time.time() - s["created_at"]
return {
"config_id": config_id, "status": "active",
"config_id": config_id, "regions": s.get("regions"), "status": "active",
"base_collected": s["base_collected"],
"sections_collected": list(s["sections_collected"]),
"sections_analyzed": list(s["sections_analyzed"]),
@@ -564,10 +589,12 @@ def _handle_cis_get_scan_status(args):
def _handle_cis_invalidate_cache(args):
config_id = args["config_id"]
if config_id in _sessions:
del _sessions[config_id]
return {"status": "cleared", "config_id": config_id}
return {"status": "not_found", "config_id": config_id}
regions = args.get("regions") or None
key = _session_key(config_id, regions)
if key in _sessions:
del _sessions[key]
return {"status": "cleared", "config_id": config_id, "regions": regions}
return {"status": "not_found", "config_id": config_id, "regions": regions}
TOOL_HANDLERS = {