admin / Synapse-Sonar

import { NextRequest, NextResponse } from "next/server";
import crypto from "crypto";
import { z } from "zod";
import { Prisma } from "@prisma/client";
import { prisma } from "@/lib/prisma";
import { evaluateSegmentation, dispatchIronNodeAlert } from "@/lib/governance";

export const runtime = "nodejs";

/**
 * POST /api/ingest/flow-logs
 *
 * Frictionless, agentless discovery endpoint. Network sensors POST observed
 * flows here with a tenant-scoped API key (Authorization: Bearer <key>).
 *
 * Responsibilities:
 *   1. Resolve tenant from the API key.
 *   2. Upsert source + destination Asset nodes for that tenant.
 *   3. Upsert the directed Connection edge (refresh lastSeen + accumulate bytes).
 *   4. Flag rogue/shadow-IT assets not present in CMDB sync.
 *   5. Run Zero-Trust evaluation and hand off violations to IronNode SOAR.
 */

const FlowSchema = z.object({
  sourceIp: z.string().ip(),
  destIp: z.string().ip(),
  sourceHostname: z.string().optional(),
  destHostname: z.string().optional(),
  protocol: z.string().optional(),
  port: z.number().int().min(0).max(65535).optional(),
  bytes: z.number().int().nonnegative().optional().default(0),
  /** IPs the sensor knows from CMDB/AD; anything outside is rogue. */
  knownInventory: z.array(z.string().ip()).optional(),
  /** Other addresses of the SOURCE host (Docker bridge / secondary NICs). The
   *  node is keyed by sourceIp (primary host IPv4); these are shown as related.
   *  Accepts an array, a single string, or null (PowerShell collapses 1-element
   *  arrays to a scalar), so collectors don't need special-casing. */
  sourceRelatedIps: z
    .union([z.array(z.string().ip()), z.string().ip()])
    .nullable()
    .optional(),
  metadata: z.record(z.unknown()).optional(),
});

/** Normalize the tolerant sourceRelatedIps shape into a clean string[]. */
function cleanRelated(ips: string[] | string | null | undefined, ownIp: string): string[] {
  if (!ips) return [];
  const arr = Array.isArray(ips) ? ips : [ips];
  return Array.from(new Set(arr.filter((ip) => ip && ip !== ownIp)));
}

const PayloadSchema = z.object({
  flows: z.array(FlowSchema).min(1).max(1000),
});

function extractBearer(req: NextRequest): string | null {
  const header = req.headers.get("authorization") ?? "";
  const match = header.match(/^Bearer\s+(.+)$/i);
  return match ? match[1].trim() : null;
}

export async function POST(req: NextRequest) {
  // --- 1. Authenticate the sensor & resolve tenant from the API key ---
  const apiKey = extractBearer(req);
  if (!apiKey) {
    return NextResponse.json({ error: "Missing bearer token" }, { status: 401 });
  }

  // The sensor key is never stored in plaintext. We persist only its SHA-256
  // hash (alongside the encrypted config blob, as config->>'ingestKeyHash') and
  // match by hashing the presented bearer token. Constant-work lookup by hash.
  const apiKeyHash = crypto.createHash("sha256").update(apiKey).digest("hex");
  const keyRow = await prisma.$queryRaw<{ tenantId: string }[]>`
    SELECT "tenantId" FROM "integration_settings"
    WHERE provider = 'NETSCAN_XI'
      AND enabled = true
      AND config->>'ingestKeyHash' = ${apiKeyHash}
    LIMIT 1;
  `;
  const tenantId = keyRow[0]?.tenantId;
  if (!tenantId) {
    return NextResponse.json({ error: "Invalid API key" }, { status: 401 });
  }

  // --- Validate payload ---
  let parsed;
  try {
    parsed = PayloadSchema.parse(await req.json());
  } catch (err) {
    return NextResponse.json(
      { error: "Invalid payload", details: (err as z.ZodError).issues ?? String(err) },
      { status: 422 }
    );
  }

  const now = new Date();
  let nodesTouched = 0;
  let edgesTouched = 0;
  let rogueFlagged = 0;
  let violations = 0;

  for (const flow of parsed.flows) {
    const known = flow.knownInventory ?? null;
    const isSourceRogue = known ? !known.includes(flow.sourceIp) : false;
    const isDestRogue = known ? !known.includes(flow.destIp) : false;
    const srcRelated = cleanRelated(flow.sourceRelatedIps, flow.sourceIp);

    // --- 2. Upsert both Asset nodes (tenant-scoped via unique [tenantId, ipAddress]) ---
    const upsertAsset = (ip: string, hostname?: string, rogue?: boolean, related?: string[]) =>
      prisma.asset.upsert({
        where: { tenantId_ipAddress: { tenantId, ipAddress: ip } },
        create: {
          tenantId,
          ipAddress: ip,
          hostname: hostname ?? null,
          isRogue: rogue ?? false,
          openPorts: flow.port ? [flow.port] : [],
          relatedIps: cleanRelated(related, ip),
          metadata: (flow.metadata ?? undefined) as Prisma.InputJsonValue | undefined,
          firstSeenAt: now,
          lastSeenAt: now,
        },
        update: {
          lastSeenAt: now,
          ...(hostname ? { hostname } : {}),
          ...(rogue ? { isRogue: true } : {}),
        },
      });

    const [source, target] = await Promise.all([
      upsertAsset(flow.sourceIp, flow.sourceHostname, isSourceRogue, srcRelated),
      upsertAsset(flow.destIp, flow.destHostname, isDestRogue),
    ]);
    nodesTouched += 2;
    if (isSourceRogue) rogueFlagged++;
    if (isDestRogue) rogueFlagged++;

    // Merge any newly-seen destination port into the target's open-port set.
    if (flow.port && !target.openPorts.includes(flow.port)) {
      await prisma.asset.update({
        where: { id: target.id },
        data: { openPorts: { set: [...target.openPorts, flow.port] } },
      });
    }

    // Merge the source host's related (Docker / secondary) addresses.
    if (srcRelated.length) {
      const merged = Array.from(new Set([...source.relatedIps, ...srcRelated]));
      if (merged.length !== source.relatedIps.length) {
        await prisma.asset.update({
          where: { id: source.id },
          data: { relatedIps: { set: merged } },
        });
      }
    }

    // --- 3. Upsert the directed Connection edge ---
    await prisma.connection.upsert({
      where: {
        tenantId_sourceAssetId_targetAssetId_port: {
          tenantId,
          sourceAssetId: source.id,
          targetAssetId: target.id,
          port: flow.port ?? 0,
        },
      },
      create: {
        tenantId,
        sourceAssetId: source.id,
        targetAssetId: target.id,
        protocol: flow.protocol ?? null,
        port: flow.port ?? 0,
        bytes: BigInt(flow.bytes),
        firstSeenAt: now,
        lastSeenAt: now,
      },
      update: {
        lastSeenAt: now,
        bytes: { increment: BigInt(flow.bytes) },
      },
    });
    edgesTouched++;

    // --- 4 & 5. Zero-Trust drift evaluation + SOAR hand-off ---
    const verdict = await evaluateSegmentation(tenantId, {
      sourceIp: flow.sourceIp,
      destIp: flow.destIp,
      port: flow.port,
    });

    if (verdict.violation || isSourceRogue || isDestRogue) {
      violations++;
      await dispatchIronNodeAlert(tenantId, {
        type: isSourceRogue || isDestRogue ? "ROGUE_ASSET" : "SEGMENTATION_VIOLATION",
        reason: verdict.reason,
        ruleId: verdict.ruleId,
        flow: {
          sourceIp: flow.sourceIp,
          destIp: flow.destIp,
          port: flow.port,
          protocol: flow.protocol,
        },
        rogue: { source: isSourceRogue, dest: isDestRogue },
      });
    }
  }

  return NextResponse.json({
    ok: true,
    summary: { nodesTouched, edgesTouched, rogueFlagged, violations },
  });
}