FortressAI is a full-stack cybersecurity platform with two operating modes in one repo:
- Demo mode: original attack-response-log simulation flow for presentations
- Demo link: https://fortress-ai-delta.vercel.app/
- Guardian mode: production-oriented blockchain wallet security for individuals and SMEs
The Guardian extension is focused on wallet drain prevention, phishing detection, risky approvals, malicious contract analysis, and real-time alerting.
- Frontend: Next.js 14 (TypeScript)
- Backend: FastAPI (Python)
- Realtime: Socket.IO + WebSocket endpoints
- Queue: Celery + Redis
- Database: PostgreSQL 16 + TimescaleDB
- Infra: Docker Compose (local), Kubernetes/EKS manifests (production)
- Recon / simulate / respond / log API pipeline
- Live dashboard at
/demo - Metrics websocket at
/ws/status
- JWT auth (
/guardian/auth/register,/guardian/auth/login) - Wallet monitoring (
/guardian/monitor-wallet) - Token approval scanner (
/guardian/scan-approvals) - Smart contract analyzer (
/guardian/analyze-contract) - Phishing URL checker (
/guardian/check-phishing) - Alert management (
/guardian/alerts, alert actions) - SIEM / IDS / firewall ingestion endpoints
- Real-time alert push over Socket.IO (
alerts:update)
Implemented with real API/service paths and cache fallbacks:
- Etherscan API (tx history and contract metadata)
- GoPlus Security API (address/token risk signals)
- MetaMask
eth-phishing-detectfeed - PhishTank feed
- ELK-compatible ingest hook (
ELASTIC_INGEST_URL, optional) - Alert providers via Celery queue:
- Discord webhook
- Telegram bot
- SendGrid email
- Twilio SMS
frontend (Next.js)
/demo -> legacy cyber demo UI
/guardian -> blockchain security dashboard
backend (FastAPI + Socket.IO ASGI app)
/scan, /simulate, /tunnel, /log
/guardian/*
/ws/status
Socket.IO event: alerts:update
services
blockchain_guardian.py -> monitor/scan/analyze/check logic
integrations.py -> Etherscan/GoPlus/phishing feeds
alerts.py -> alert persistence + SIEM + realtime emit
notifications.py -> provider fanout (Discord/Telegram/SendGrid/Twilio)
tasks (Celery)
guardian scans
alert delivery with retries
.
├── backend/
│ ├── app/
│ │ ├── main.py
│ │ ├── realtime.py
│ │ ├── routers/
│ │ │ ├── api.py
│ │ │ └── guardian.py
│ │ ├── services/
│ │ ├── tasks.py
│ │ └── ws/socket.py
│ └── tests/
├── frontend/
│ ├── app/
│ │ ├── demo/page.tsx
│ │ └── guardian/page.tsx
│ └── lib/
├── infra/
│ └── k8s/
├── scripts/
└── docker-compose.yml
- GitHub-safe template:
.env.example - Local secrets file (git-ignored):
.env - Kubernetes secret template:
infra/k8s/guardian-secrets.example.yaml - Real local
.envpath in this workspace:/workspaces/FortressAI/.env
Create local env from template:
cp .env.example .envdocker compose up --buildServices:
- Frontend:
http://localhost:3000 - Backend:
http://localhost:8000 - Postgres:
localhost:5432 - Redis:
localhost:6379
- Demo:
http://localhost:3000/demo - Guardian:
http://localhost:3000/guardian
On backend startup, FortressAI seeds two default Guardian users if they do not already exist:
demo.hk@fortressai.local/Fortress123!(regionHK)demo.ke@fortressai.local/Fortress123!
DATABASE_URLREDIS_URLJWT_SECRET_KEY
ETHERSCAN_API_KEYETHERSCAN_BASE_URL(optional override)GOPLUS_BASE_URL(optional override)METAMASK_PHISHING_URL(optional override)PHISHTANK_FEED_URL(optional override)
ELASTIC_INGEST_URL(optional)AWS_SECURITY_HUB_ENABLED(optional)AWS_REGION(optional)
DISCORD_WEBHOOK_URLTELEGRAM_BOT_TOKENTELEGRAM_CHAT_IDSENDGRID_API_KEYALERT_EMAIL_FROMALERT_EMAIL_TOTWILIO_ACCOUNT_SIDTWILIO_AUTH_TOKENTWILIO_FROM_NUMBERTWILIO_TO_NUMBER
Base URL: http://localhost:8000
POST /scanPOST /simulatePOST /tunnelPOST /logGET /status
POST /guardian/auth/registerPOST /guardian/auth/loginGET /guardian/auth/mePOST /guardian/monitor-walletPOST /guardian/scan-approvalsPOST /guardian/analyze-contractPOST /guardian/check-phishingGET /guardian/alertsPOST /guardian/alerts/{alert_id}/actionPOST /guardian/ingest/siemPOST /guardian/ingest/idsPOST /guardian/ingest/firewallGET /guardian/startup-checkGET /guardian/telegram/chat-idGET /guardian/telegram/recent-updatesPOST /guardian/telegram/discover-chat-idPOST /guardian/telegram/store-chat-id
WS /ws/status(legacy metrics)- Socket.IO namespace default at
/socket.io - Socket.IO auth:
auth.token = <JWT> - Event emitted by backend:
alerts:update
Alert fanout is queued through Celery task:
tasks.alert.deliver
Retry strategy:
- Exponential backoff
- Jitter enabled
- Max retries: 5
cd frontend
npm test -- --watch=false
npm run buildpip install -r backend/requirements.txt
pytest -q backend/testsRun unit-focused tests only:
pytest -q backend/tests -m "not integration"Run integration tests only:
pytest -q backend/tests -m integrationDiscover chat IDs from recent updates and persist one to .env:
python scripts/telegram_chat_id_helper.py --save-envChoose a specific chat ID instead of latest:
python scripts/telegram_chat_id_helper.py --choose <chat_id> --save-envCurrent backend tests include:
- Guardian route behavior
- Risk-scoring paths
- Socket.IO JWT connect +
alerts:updateemission flow - Network-level Socket.IO
AsyncClientintegration test against live Uvicorn app
Use:
scripts/deploy_eks_5min.shRequired env:
AWS_ACCOUNT_ID
Optional env:
AWS_REGIONIMAGE_TAGK8S_NAMESPACEECR_REPO_BACKENDECR_REPO_FRONTEND
Use infra/k8s/guardian-secrets.example.yaml as the starting point for cluster secrets.
- Private keys are never stored.
- Wallet monitoring is consent-based via registered wallet addresses.
- Sensitive integration credentials are expected from env or Kubernetes secrets.
- API middleware includes request rate limiting.
- Some provider features require API keys and may fallback to cache/partial signals without them.
- Contract analysis currently combines explorer metadata + risk feeds + heuristic checks, not full symbolic execution in cluster by default.
- Legacy demo tests may reference older UI labels; Guardian tests are separate.