Files
trustos/backend/seed.py
drjones 631a6b4147 Fix code issues and add missing documentation
- Fix duplicate tenant_id parameter in seed.py (line 148)
- Add security warning to SECRET_KEY in .env.example
- Create comprehensive README.md with setup instructions
- Add Alembic configuration files (alembic.ini, env.py, script.py.mako)
- Create initial database migration for all tables
- Document project structure, features, and deployment checklist
2026-07-06 02:58:08 +00:00

232 lines
13 KiB
Python

"""
Seed script — creates the demo tenant, users, assets, findings, and risk score history.
Run with: python seed.py
"""
import asyncio
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from datetime import datetime, timedelta
from app.db.session import engine, AsyncSessionLocal
from app.models.models import (
Base, Tenant, User, UserRole, Asset, AssetType, Finding,
FindingSeverity, FindingStatus, FindingCategory, RiskScore,
AuthorizedAsset, Executive
)
from app.core.security import hash_password
DEMO_FINDINGS = [
{
"title": "Internet-accessible admin panel with no authentication",
"severity": FindingSeverity.critical,
"category": FindingCategory.external_exposure,
"technical_description": "The administrative interface at admin.acmecorp.io is exposed to the internet without any authentication requirement. An unauthenticated attacker can access full system configuration.",
"affected_component": "admin.acmecorp.io:443",
"ai_summary": "Your internal admin portal is accessible to anyone on the internet without a password.",
"ai_business_impact": "An attacker could gain full control of your platform, access all customer data, and disrupt operations — potentially within minutes of discovery.",
"ai_impact_level": "Critical",
"ai_remediation_steps": "1. Immediately restrict admin portal access to VPN/office IPs only.\n2. Add multi-factor authentication.\n3. Audit access logs for any unauthorized access.\n4. Review what data the portal can access.",
"ai_fix_priority": "urgent",
"is_top_risk": True,
"source": "manual",
},
{
"title": "5 executive email accounts found in breach database",
"severity": FindingSeverity.high,
"category": FindingCategory.credential_exposure,
"technical_description": "HaveIBeenPwned confirmed that 5 corporate email addresses belonging to C-suite executives (including CEO and CFO) appear in the Stealer Logs breach dataset with associated plaintext passwords.",
"affected_component": "Corporate email / M365",
"ai_summary": "Passwords for your senior leadership email accounts have been leaked and are available on the dark web.",
"ai_business_impact": "Attackers could use these credentials to access email, cloud systems, and sensitive financial data — enabling targeted phishing, wire fraud, or full account takeover.",
"ai_impact_level": "High",
"ai_remediation_steps": "1. Force immediate password reset for all affected accounts.\n2. Enable MFA on all executive accounts if not already active.\n3. Review recent email forwarding rules and login activity.\n4. Brief executives on spear-phishing risk.",
"ai_fix_priority": "urgent",
"is_top_risk": True,
"source": "hibp",
},
{
"title": "S3 storage bucket publicly accessible — contains customer files",
"severity": FindingSeverity.critical,
"category": FindingCategory.cloud_posture,
"technical_description": "AWS S3 bucket 'acme-customer-uploads-prod' has public read access enabled. The bucket contains customer-uploaded documents including contracts and personally identifiable information.",
"cve_id": None,
"affected_component": "AWS S3: acme-customer-uploads-prod",
"ai_summary": "A cloud storage location containing your customer files is publicly accessible — anyone on the internet can read them.",
"ai_business_impact": "This constitutes a data breach. Exposure of customer PII could trigger regulatory penalties, customer notification obligations, and severe damage to customer trust.",
"ai_impact_level": "Critical",
"ai_remediation_steps": "1. Immediately set bucket ACL to private.\n2. Audit which files were stored there and since when.\n3. Determine if any unauthorized access occurred via S3 access logs.\n4. Notify legal counsel — this may trigger breach notification requirements.\n5. Review all other S3 buckets for similar misconfigurations.",
"ai_fix_priority": "urgent",
"is_top_risk": True,
"source": "cloud_scan",
},
{
"title": "SSL/TLS certificate expiring in 12 days",
"severity": FindingSeverity.medium,
"category": FindingCategory.external_exposure,
"technical_description": "The TLS certificate for api.acmecorp.io expires in 12 days. Expiration will cause browser security warnings and API failures for all customers.",
"affected_component": "api.acmecorp.io",
"ai_summary": "Your API security certificate is about to expire — this will break your application for customers in less than two weeks.",
"ai_business_impact": "Customers will see security errors and be unable to use your product, resulting in direct revenue impact and support escalation.",
"ai_impact_level": "Medium",
"ai_remediation_steps": "1. Renew the certificate immediately via your certificate authority.\n2. Set up auto-renewal to prevent this in the future.\n3. Verify renewal across all sub-domains.",
"ai_fix_priority": "soon",
"is_top_risk": False,
"source": "cert_monitor",
},
{
"title": "CEO LinkedIn profile reveals unreported board membership and home city",
"severity": FindingSeverity.medium,
"category": FindingCategory.digital_footprint,
"technical_description": "OSINT analysis of the CEO's public LinkedIn profile reveals home city, secondary board membership not disclosed in company materials, and personal email address listed publicly. This information could be used for targeted social engineering.",
"affected_component": "Executive digital footprint",
"ai_summary": "Publicly available information about your CEO can help attackers craft convincing impersonation or social engineering attacks.",
"ai_business_impact": "Sophisticated attackers use personal information to craft targeted phishing attacks, impersonate executives in wire transfer fraud, or manipulate employees.",
"ai_impact_level": "Medium",
"ai_remediation_steps": "1. Review and reduce personal information on professional profiles.\n2. Brief CEO on executive-targeted social engineering tactics.\n3. Remove personal email from public profiles.\n4. Enable enhanced privacy settings on all platforms.",
"ai_fix_priority": "soon",
"is_top_risk": False,
"source": "manual",
},
{
"title": "Web application missing security headers (HSTS, CSP, X-Frame-Options)",
"severity": FindingSeverity.medium,
"category": FindingCategory.web_application,
"technical_description": "The main web application at app.acmecorp.io is missing HTTP Strict Transport Security (HSTS), Content Security Policy (CSP), and X-Frame-Options headers. This increases susceptibility to clickjacking and protocol downgrade attacks.",
"affected_component": "app.acmecorp.io",
"ai_summary": "Your web application is missing basic browser security protections that prevent common attacks.",
"ai_business_impact": "Users could be redirected to malicious sites or have their sessions hijacked through man-in-the-middle attacks, exposing customer data.",
"ai_impact_level": "Medium",
"ai_remediation_steps": "1. Add HSTS header with a minimum 1-year max-age.\n2. Implement a Content Security Policy.\n3. Add X-Frame-Options: DENY.\n4. Test changes in a staging environment first.",
"ai_fix_priority": "soon",
"is_top_risk": False,
"source": "scanner",
},
]
async def seed():
# Ensure tables exist
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with AsyncSessionLocal() as db:
# ── Demo Tenant ──────────────────────────────────────────────────
tenant = Tenant(
id="acme-corp-demo-001",
name="Acme Corp",
slug="acme-corp",
industry="SaaS",
size_range="100-300",
contact_email="it@acmecorp.io",
)
db.add(tenant)
# ── Users ────────────────────────────────────────────────────────
users = [
User(
tenant_id="acme-corp-demo-001",
email="executive@acmecorp.io",
hashed_password=hash_password("TrustOS2024!"),
full_name="Sarah Chen (CEO)",
role=UserRole.executive,
),
User(
tenant_id="acme-corp-demo-001",
email="it@acmecorp.io",
hashed_password=hash_password("TrustOS2024!"),
full_name="Marcus Johnson (IT Director)",
role=UserRole.it_admin,
),
User(
tenant_id="acme-corp-demo-001",
email="admin@trustos.com",
hashed_password=hash_password("TrustOS-Admin-2024!"),
full_name="TrustOS Admin",
role=UserRole.trustos_admin,
),
]
for u in users:
db.add(u)
# ── Authorized Assets (scope-lock) ───────────────────────────────
auth_assets = [
AuthorizedAsset(
tenant_id="acme-corp-demo-001",
value="acmecorp.io",
asset_type=AssetType.domain,
description="Primary company domain",
authorized_by="it@acmecorp.io",
),
AuthorizedAsset(
tenant_id="acme-corp-demo-001",
value="*.acmecorp.io",
asset_type=AssetType.domain,
description="All subdomains",
authorized_by="it@acmecorp.io",
),
]
for a in auth_assets:
db.add(a)
# ── Assets ────────────────────────────────────────────────────────
assets = [
Asset(tenant_id="acme-corp-demo-001", name="Main Website", asset_type=AssetType.domain, value="acmecorp.io"),
Asset(tenant_id="acme-corp-demo-001", name="Admin Portal", asset_type=AssetType.web_application, value="admin.acmecorp.io"),
Asset(tenant_id="acme-corp-demo-001", name="Customer API", asset_type=AssetType.api_endpoint, value="api.acmecorp.io"),
Asset(tenant_id="acme-corp-demo-001", name="Customer Upload Bucket", asset_type=AssetType.cloud_resource, value="s3://acme-customer-uploads-prod"),
]
for a in assets:
db.add(a)
# ── Executives ───────────────────────────────────────────────────
execs = [
Executive(tenant_id="acme-corp-demo-001", full_name="Sarah Chen", title="CEO", corporate_email="sarah.chen@acmecorp.io"),
Executive(tenant_id="acme-corp-demo-001", full_name="David Park", title="CFO", corporate_email="david.park@acmecorp.io"),
]
for e in execs:
db.add(e)
await db.flush()
# ── Findings ──────────────────────────────────────────────────────
for f_data in DEMO_FINDINGS:
finding = Finding(tenant_id="acme-corp-demo-001", ai_generated_at=datetime.utcnow(), **f_data)
db.add(finding)
await db.flush()
# ── Historical Risk Scores (90 days) ──────────────────────────────
base_score = 58.0
for i in range(90, 0, -1):
delta = (90 - i) * 0.35
score = min(100, round(base_score + delta + (i % 3 - 1) * 0.5, 1))
rs = RiskScore(
tenant_id="acme-corp-demo-001",
score_date=datetime.utcnow() - timedelta(days=i),
overall_score=score,
score_identity=min(100, score + 5),
score_cloud=max(0, score - 10),
score_network=score,
score_web=score + 2,
score_credential=max(0, score - 8),
score_digital_footprint=score + 3,
score_third_party=score + 1,
critical_count=3 if i > 30 else 2,
high_count=2,
medium_count=3,
low_count=1,
)
db.add(rs)
await db.commit()
print("✓ Seed complete — demo tenant 'Acme Corp' created")
print(" executive@acmecorp.io / TrustOS2024!")
print(" it@acmecorp.io / TrustOS2024!")
print(" admin@trustos.com / TrustOS-Admin-2024!")
if __name__ == "__main__":
asyncio.run(seed())