admin / Syanpse-Vanguard

"""End-to-end smoke test with no external services (memory broker + DuckDB).

Exercises the real code paths for Milestones 1-3:
  1. storage: schema init + bulk write + parameterized search
  2. ingest:  agent enrollment (dev bearer) + broker push
  3. pipeline: normalizer parse + durable write + ack
  4. query:   RBAC-guarded, tenant-isolated search
  5. rbac:    role matrix enforcement

Run:  python scripts/smoke.py
"""
from __future__ import annotations

import asyncio
import json
import os
import tempfile

# Configure a self-contained environment BEFORE importing sv.*
_tmp = tempfile.mkdtemp(prefix="sv-smoke-")
os.environ.update(
    SV_STORAGE_BACKEND="duckdb",
    SV_DUCKDB_PATH=os.path.join(_tmp, "smoke.duckdb"),
    SV_COLD_DIR=os.path.join(_tmp, "cold"),
    SV_BROKER="memory",
    SV_MTLS_ENABLED="false",
    SV_VAULT_KEY="11" * 32,
    SV_JWT_SECRET="smoke-secret",
    SV_ENROLLMENT_TOKEN="smoke-enroll",
    SV_BOOTSTRAP_ADMIN_USER="admin",
    SV_BOOTSTRAP_ADMIN_PASSWORD="admin-pw",
)

from sv.auth.rbac import Permission, Role, can                       # noqa: E402
from sv.auth.session import issue_token                             # noqa: E402
from sv.auth.users import get_user_by_username, verify_password     # noqa: E402
from sv.bootstrap import shutdown, startup                          # noqa: E402
from sv.ingest import registry                                      # noqa: E402
from sv.models.agent import AgentRegisterRequest                    # noqa: E402
from sv.pipeline.normalizer import _parse_lines                     # noqa: E402
from sv.query.builder import SearchRequest, build_search            # noqa: E402
from sv.config import DEFAULT_TENANT_ID                             # noqa: E402

PASS, FAIL = "\033[92mPASS\033[0m", "\033[91mFAIL\033[0m"
_ok = True


def check(name: str, cond: bool) -> None:
    global _ok
    _ok = _ok and cond
    print(f"  [{PASS if cond else FAIL}] {name}")


async def main() -> int:
    engine, broker = await startup(with_broker=True)
    try:
        # 1. bootstrap admin seeded
        admin = await get_user_by_username(engine, "admin")
        check("bootstrap admin seeded", admin is not None)
        check("admin password verifies", verify_password("admin-pw", admin["password_hash"]))
        check("admin is TENANT_ADMIN", admin["role"] == "TENANT_ADMIN")

        # 2. RBAC matrix
        check("auditor can read logs", can(Role.READ_ONLY_VIEWER, Permission.LOG_READ))
        check("auditor CANNOT triage", not can(Role.READ_ONLY_VIEWER, Permission.ALERT_TRIAGE))
        check("analyst can triage", can(Role.SECURITY_ANALYST, Permission.ALERT_TRIAGE))
        check("analyst CANNOT manage users", not can(Role.SECURITY_ANALYST, Permission.USER_MANAGE))
        check("admin can manage agents", can(Role.TENANT_ADMIN, Permission.AGENT_MANAGE))

        # 3. JWT round-trips
        tok = issue_token(admin["id"], "admin", Role.TENANT_ADMIN, DEFAULT_TENANT_ID)
        check("jwt issued", isinstance(tok, str) and tok.count(".") == 2)

        # 4. agent enrollment (dev bearer)
        req = AgentRegisterRequest(hostname="testhost", platform="linux",
                                   agent_version="0.1.0", channels=["journald"])
        agent_id, token = await registry.enroll(engine, req, cert_cn=None)
        check("agent enrolled with dev token", token is not None)
        found = await registry.get_agent(engine, agent_id)
        check("agent retrievable", found is not None and not found["revoked"])

        # 5. ingest -> broker -> normalizer -> storage
        lines = [
            json.dumps({"source_type": "winlog", "host": "testhost",
                        "message": "An account was successfully logged on",
                        "fields": {"EventID": 4624, "Channel": "Security",
                                   "Level": 4, "TargetUserName": "alice",
                                   "Keywords": "Audit Success"}}),
            json.dumps({"source_type": "journald", "host": "testhost",
                        "message": "sshd accepted password for bob",
                        "fields": {"PRIORITY": "6", "_COMM": "sshd"}}),
            "{ this is not valid json }",   # must be tolerated & dropped
        ]
        await broker.push_raw(agent_id, seq=2, payload="\n".join(lines).encode())

        msgs = await broker.read_group("smoke", count=10, block_ms=500)
        check("broker delivered 1 message", len(msgs) == 1)
        events = []
        for m in msgs:
            events.extend(_parse_lines(m.payload))
            await broker.ack(m.msg_id)
        check("2 valid events parsed (1 bad dropped)", len(events) == 2)
        written = await engine.write_batch(events)
        check("2 events written to storage", written == 2)

        # 6. normalization correctness
        win = next(e for e in events if e.source_type == "winlog")
        check("winlog EventID parsed", win.event_id == 4624)
        check("winlog outcome=success", win.outcome == "success")
        check("winlog user parsed", win.user_name == "alice")

        # 7. RBAC- & tenant-scoped search
        q = build_search(SearchRequest(equals={"source_type": "winlog"}, limit=10),
                         tenant_id=DEFAULT_TENANT_ID)
        table = await engine.execute(q)
        check("search returns the winlog event", table.num_rows == 1)
        check("tenant filter present in SQL", "tenant_id = ?" in q.text)

        # 8. filter allowlist rejects unknown fields (injection guard)
        try:
            build_search(SearchRequest(equals={"evil; DROP TABLE events": 1}),
                         tenant_id=DEFAULT_TENANT_ID)
            check("unknown filter rejected", False)
        except ValueError:
            check("unknown filter rejected", True)

    finally:
        await shutdown(engine, broker)

    print()
    print("RESULT:", "ALL PASS" if _ok else "FAILURES PRESENT")
    return 0 if _ok else 1


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))