Docs

Appearance

UBA System — Baseline Builder and Longitudinal Analysis Agent ​

Overview ​

User Behavior Analytics (UBA) is a two-component system that detects behavioral drift across users, devices, and service accounts. It is designed to catch threats that evade signature-based detection: compromised credentials used in a "normal" way, insider threats that operate within policy, and slow-burn lateral movement that looks like noise in any single four-hour window.

Component 1 — Baseline Builder (job-uba-baseline): Runs weekly. Queries 30 days of historical data from Log Analytics Workspace (LAW) cirius-logging-law-central, computes p50/p95 and standard deviation for each metric per entity, and upserts the results into PostgreSQL psql-secops-prod. This is the statistical reference point everything else compares against.

Component 2 — Longitudinal UBA Agent (job-uba-agent): Runs daily inside the Container Apps Job ca-secops-prod. Pulls baselines from PostgreSQL, queries the last 24 hours from LAW, computes drift scores, and posts anomalies to the SecOps platform at soc.bedrockcybersecurity.org/api/incidents.

Neither component uses LLM enrichment by default. Azure OpenAI (gpt-4o at cirius-openai-prod.openai.azure.com) is available for anomaly explanation when the flag is set, but detection logic does not depend on it.

Infrastructure Reference ​

ComponentResourceNotes
Log Analyticscirius-logging-law-centralWorkspace ID 5d76d1f2
PostgreSQLpsql-secops-prodAzure Database for PostgreSQL Flexible Server
SecOps platformsecops.bedrockcybersecurity.orgFastAPI + PostgreSQL, Container App ca-soc-prod
Azure OpenAIcirius-openai-prod.openai.azure.comgpt-4o, optional enrichment
Container Apps Jobca-secops-prodResource group rg-logging-logs
Key Vaultcirius-openai-kv-prodAuth credentials and connection strings
Auth (Azure)DefaultAzureCredentialManaged identity on both jobs
Auth (PostgreSQL)Key Vault secretRetrieved at runtime, not in environment

UBA jobs must have their managed identity assigned the following roles:

  • Monitoring Reader on cirius-logging-law-central — for LAW queries
  • Key Vault Secrets User on cirius-openai-kv-prod — for PostgreSQL connection string
  • No direct PostgreSQL IAM role — connection string carries credentials

Database Schema ​

All UBA tables live in the uba schema on psql-secops-prod. Run these once during initial deployment. Alembic manages subsequent migrations.

sql
-- Create schema
CREATE SCHEMA IF NOT EXISTS uba;

-- uba.entities: one row per tracked entity
-- entity_type: 'user' | 'device' | 'service_account'
-- entity_id: UPN for users, hostname for devices, SAM account for service accounts
CREATE TABLE IF NOT EXISTS uba.entities (
    id                  SERIAL PRIMARY KEY,
    entity_type         VARCHAR(32)  NOT NULL,
    entity_id           VARCHAR(256) NOT NULL,
    display_name        VARCHAR(256),
    department          VARCHAR(128),
    first_seen          TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    last_activity       TIMESTAMPTZ,
    baseline_valid      BOOLEAN      NOT NULL DEFAULT FALSE,
    baseline_days       INTEGER      NOT NULL DEFAULT 0,
    notes               TEXT,
    UNIQUE (entity_type, entity_id)
);

CREATE INDEX idx_entities_type ON uba.entities (entity_type);
CREATE INDEX idx_entities_id   ON uba.entities (entity_id);

-- uba.baselines: statistical baseline per entity per metric
-- computed_at: when this baseline was last built
-- window_days: how many days of data went into this calculation
-- p50, p95: 50th and 95th percentile of daily values
-- sigma: standard deviation of daily values
-- sample_count: number of daily data points used
-- suppressed: if true, skip anomaly detection for this metric (e.g. role change)
CREATE TABLE IF NOT EXISTS uba.baselines (
    id              SERIAL PRIMARY KEY,
    entity_id       INTEGER      NOT NULL REFERENCES uba.entities(id) ON DELETE CASCADE,
    metric          VARCHAR(64)  NOT NULL,
    computed_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    window_days     INTEGER      NOT NULL DEFAULT 30,
    p50             FLOAT        NOT NULL DEFAULT 0,
    p95             FLOAT        NOT NULL DEFAULT 0,
    sigma           FLOAT        NOT NULL DEFAULT 0,
    sample_count    INTEGER      NOT NULL DEFAULT 0,
    suppressed      BOOLEAN      NOT NULL DEFAULT FALSE,
    suppress_reason TEXT,
    suppress_until  TIMESTAMPTZ,
    UNIQUE (entity_id, metric)
);

CREATE INDEX idx_baselines_entity  ON uba.baselines (entity_id);
CREATE INDEX idx_baselines_metric  ON uba.baselines (metric);
CREATE INDEX idx_baselines_suppressed ON uba.baselines (suppressed, suppress_until);

-- uba.daily_snapshots: raw daily aggregate per entity per metric
-- date: UTC date of the measurement
-- value: raw computed value for that day (count, bytes, rate, etc.)
-- Used both by the baseline builder (reads 30-day history) and as a rolling log
CREATE TABLE IF NOT EXISTS uba.daily_snapshots (
    id          BIGSERIAL    PRIMARY KEY,
    entity_id   INTEGER      NOT NULL REFERENCES uba.entities(id) ON DELETE CASCADE,
    metric      VARCHAR(64)  NOT NULL,
    date        DATE         NOT NULL,
    value       FLOAT        NOT NULL,
    source      VARCHAR(32)  NOT NULL DEFAULT 'law',
    UNIQUE (entity_id, metric, date)
);

CREATE INDEX idx_snapshots_entity ON uba.daily_snapshots (entity_id);
CREATE INDEX idx_snapshots_date   ON uba.daily_snapshots (date DESC);
CREATE INDEX idx_snapshots_em     ON uba.daily_snapshots (entity_id, metric, date DESC);

-- uba.anomalies: every drift event detected by the longitudinal agent
-- ioc_key: deduplication key, format UBA:{entity_type}:{entity_id}:{metric}:{date}
-- severity: CRITICAL | HIGH | MEDIUM | LOW
-- observed_value: the 24h value that triggered the anomaly
-- p95: p95 from the baseline at detection time (snapshot)
-- sigma: sigma from the baseline at detection time
-- z_score: (observed_value - p95) / sigma — how many sigmas above p95
-- incident_id: SecOps incident ID after successful post, null if not yet posted
-- explanation: LLM-generated text if enrichment is enabled, else null
CREATE TABLE IF NOT EXISTS uba.anomalies (
    id              BIGSERIAL    PRIMARY KEY,
    entity_id       INTEGER      NOT NULL REFERENCES uba.entities(id) ON DELETE CASCADE,
    metric          VARCHAR(64)  NOT NULL,
    detected_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    analysis_date   DATE         NOT NULL,
    severity        VARCHAR(16)  NOT NULL,
    observed_value  FLOAT        NOT NULL,
    baseline_p50    FLOAT        NOT NULL,
    baseline_p95    FLOAT        NOT NULL,
    baseline_sigma  FLOAT        NOT NULL,
    z_score         FLOAT,
    ioc_key         VARCHAR(512) NOT NULL UNIQUE,
    incident_id     INTEGER,
    posted_at       TIMESTAMPTZ,
    explanation     TEXT,
    false_positive  BOOLEAN      NOT NULL DEFAULT FALSE,
    fp_reason       TEXT,
    fp_marked_at    TIMESTAMPTZ,
    fp_marked_by    VARCHAR(128)
);

CREATE INDEX idx_anomalies_entity   ON uba.anomalies (entity_id);
CREATE INDEX idx_anomalies_date     ON uba.anomalies (analysis_date DESC);
CREATE INDEX idx_anomalies_severity ON uba.anomalies (severity);
CREATE INDEX idx_anomalies_ioc      ON uba.anomalies (ioc_key);
CREATE INDEX idx_anomalies_fp       ON uba.anomalies (false_positive) WHERE false_positive = TRUE;

Retention: Daily snapshots older than 90 days should be pruned. Anomaly records are kept for 6 years per HIPAA audit requirements. See HIPAA Considerations for what is and is not logged.

Baseline Metrics Reference ​

User-Level Metrics ​

Metric NameDescriptionData SourceUnit
user_logon_count_dailyTotal successful sign-ins per daySignInLogscount
user_unique_src_ips_dailyDistinct source IPs per daySignInLogscount
user_offhour_logon_rateFraction of logons outside 07:00–19:00 localSignInLogsratio 0–1
user_failed_logon_rateFailed/(failed+successful) logons per daySignInLogsratio 0–1
user_pim_activations_dailyPrivileged role activations per dayAuditLogscount
user_mfa_challenge_rateMFA challenges per successful logonSignInLogsratio 0–1
user_m365_teams_events_dailyTeams events (messages, calls) per dayOfficeActivitycount
user_m365_sharepoint_events_dailySharePoint file operations per dayOfficeActivitycount
user_m365_exchange_events_dailyExchange send/receive events per dayOfficeActivitycount

Device-Level Metrics ​

Metric NameDescriptionData SourceUnit
device_bytes_sent_dailyOutbound network bytes per dayPalo Alto CEF / CommonSecurityLogbytes
device_bytes_recv_dailyInbound network bytes per dayPalo Alto CEF / CommonSecurityLogbytes
device_unique_dst_ips_dailyDistinct destination IPs per dayCommonSecurityLogcount
device_process_exec_dailyProcess creation events (4688) per daySecurityEventcount
device_new_services_dailyNew service installs (7045) per daySecurityEventcount
device_new_tasks_dailyNew scheduled tasks (4698) per daySecurityEventcount
device_new_dst_ips_dailyOutbound connections to IPs not seen in prior 30dCommonSecurityLogcount

Service Account-Level Metrics ​

Metric NameDescriptionData SourceUnit
svc_unique_src_hosts_dailyDistinct source hostnames per daySecurityEventcount
svc_interactive_logon_dailyInteractive logon attempts (type 2/10) per daySecurityEventcount
svc_offhour_activity_rateFraction of activity outside 06:00–22:00SecurityEventratio 0–1

KQL Queries ​

These are the canonical queries for each metric category. The baseline builder and longitudinal agent both use these queries — the difference is only the time window (startofday(ago(30d)) vs startofday(ago(1d))).

User Sign-In Behavior ​

kql
// user_logon_count_daily, user_unique_src_ips_daily, user_offhour_logon_rate,
// user_failed_logon_rate, user_mfa_challenge_rate
//
// Replace START_DATE and END_DATE with your window bounds (inclusive)
SignInLogs
| where TimeGenerated between (datetime(START_DATE) .. datetime(END_DATE))
| where UserPrincipalName != ""
| extend logon_date = startofday(TimeGenerated)
| extend is_success = ResultType == 0
| extend is_failure = ResultType != 0 and ResultType != 50126  // exclude expected MFA
| extend is_offhour = hourofday(TimeGenerated) < 7 or hourofday(TimeGenerated) >= 19
| extend is_mfa = AuthenticationRequirement == "multiFactorAuthentication"
| summarize
    total_logons      = count(),
    successful_logons = countif(is_success),
    failed_logons     = countif(is_failure),
    offhour_logons    = countif(is_offhour and is_success),
    mfa_challenges    = countif(is_mfa),
    unique_src_ips    = dcount(IPAddress)
    by UserPrincipalName, logon_date
| extend
    failed_rate   = toreal(failed_logons) / toreal(total_logons + 1),
    offhour_rate  = toreal(offhour_logons) / toreal(successful_logons + 1),
    mfa_rate      = toreal(mfa_challenges) / toreal(successful_logons + 1)
| project
    entity_id     = toupper(UserPrincipalName),
    date          = logon_date,
    user_logon_count_daily          = successful_logons,
    user_unique_src_ips_daily       = unique_src_ips,
    user_offhour_logon_rate         = offhour_rate,
    user_failed_logon_rate          = failed_rate,
    user_mfa_challenge_rate         = mfa_rate
| order by entity_id asc, date asc

Privileged Role Activations (PIM) ​

kql
// user_pim_activations_daily
AuditLogs
| where TimeGenerated between (datetime(START_DATE) .. datetime(END_DATE))
| where OperationName has "Add member to role completed (PIM activation)"
    or OperationName has "Activate"
| where Category == "RoleManagement"
| extend logon_date = startofday(TimeGenerated)
| extend upn = tostring(TargetResources[0].userPrincipalName)
| where upn != ""
| summarize
    pim_activations = count()
    by entity_id = toupper(upn), date = logon_date
| order by entity_id asc, date asc

M365 Application Usage ​

kql
// user_m365_teams_events_daily, user_m365_sharepoint_events_daily,
// user_m365_exchange_events_daily
OfficeActivity
| where TimeGenerated between (datetime(START_DATE) .. datetime(END_DATE))
| where UserId != "" and UserId != "app@sharepoint"
| extend logon_date = startofday(TimeGenerated)
| extend workload = RecordType
| summarize
    teams_events      = countif(workload == "MicrosoftTeams"),
    sharepoint_events = countif(workload == "SharePoint" or workload == "OneDrive"),
    exchange_events   = countif(workload == "ExchangeAdmin" or workload == "ExchangeItem")
    by entity_id = toupper(UserId), date = logon_date
| order by entity_id asc, date asc

Device Network Traffic (Palo Alto) ​

kql
// device_bytes_sent_daily, device_bytes_recv_daily,
// device_unique_dst_ips_daily, device_new_dst_ips_daily
//
// CommonSecurityLog carries Palo Alto CEF after CEF->LAW forwarding is configured.
// SentBytes / ReceivedBytes are in the raw CEF fields.
// For device_new_dst_ips_daily: compare DestinationIP against IPs seen in prior 30 days.
CommonSecurityLog
| where TimeGenerated between (datetime(START_DATE) .. datetime(END_DATE))
| where DeviceVendor == "Palo Alto Networks"
| where DeviceEventClassID == "TRAFFIC"
| where SourceHostName != "" and DestinationIP != ""
| extend logon_date = startofday(TimeGenerated)
| extend src_host = toupper(tostring(split(SourceHostName, ".")[0]))
| summarize
    bytes_sent       = sum(SentBytes),
    bytes_recv       = sum(ReceivedBytes),
    unique_dst_ips   = dcount(DestinationIP),
    all_dst_ips      = make_set(DestinationIP, 10000)
    by entity_id = src_host, date = logon_date
| order by entity_id asc, date asc

// For new destination detection, the Python layer computes device_new_dst_ips_daily
// by comparing all_dst_ips against the union of ips seen in days -31 to -1.
// Do not attempt this in KQL across a 30-day window — it's expensive. Pull daily
// sets and do set arithmetic in Python.

Process Execution (EventID 4688) ​

kql
// device_process_exec_daily
SecurityEvent
| where TimeGenerated between (datetime(START_DATE) .. datetime(END_DATE))
| where EventID == 4688
| where Computer != ""
| extend logon_date = startofday(TimeGenerated)
| extend host = toupper(tostring(split(Computer, ".")[0]))
| summarize
    process_exec_count = count()
    by entity_id = host, date = logon_date
| order by entity_id asc, date asc

Service and Scheduled Task Creation (7045 / 4698) ​

kql
// device_new_services_daily, device_new_tasks_daily
SecurityEvent
| where TimeGenerated between (datetime(START_DATE) .. datetime(END_DATE))
| where EventID in (7045, 4698)
| where Computer != ""
| extend logon_date = startofday(TimeGenerated)
| extend host = toupper(tostring(split(Computer, ".")[0]))
| summarize
    new_services = countif(EventID == 7045),
    new_tasks    = countif(EventID == 4698)
    by entity_id = host, date = logon_date
| order by entity_id asc, date asc

Service Account Activity ​

kql
// svc_unique_src_hosts_daily, svc_interactive_logon_daily, svc_offhour_activity_rate
//
// Service accounts are identified by the naming convention: accounts with a name
// starting with "svc-" or "sa-" in SecurityEvent, or via a reference list in PostgreSQL.
// Use the PostgreSQL reference list — it's more reliable than pattern matching.
SecurityEvent
| where TimeGenerated between (datetime(START_DATE) .. datetime(END_DATE))
| where EventID in (4624, 4625, 4648)
| where TargetUserName != "" and TargetDomainName != ""
| extend logon_date = startofday(TimeGenerated)
| extend is_interactive = LogonType in (2, 10)
| extend is_offhour = hourofday(TimeGenerated) < 6 or hourofday(TimeGenerated) >= 22
| extend src_host = toupper(tostring(split(WorkstationName, ".")[0]))
| summarize
    unique_src_hosts     = dcount(src_host),
    interactive_logons   = countif(is_interactive),
    total_events         = count(),
    offhour_events       = countif(is_offhour)
    by entity_id = strcat(toupper(TargetDomainName), "\\", toupper(TargetUserName)),
       date = logon_date
| extend offhour_rate = toreal(offhour_events) / toreal(total_events + 1)
| order by entity_id asc, date asc

Baseline Builder — Python Implementation ​

The baseline builder runs as Container Apps Job job-uba-baseline on a weekly schedule (Sundays at 02:00 UTC). Its sole job is to query LAW for the last 30 days, compute statistics per entity per metric, and upsert to PostgreSQL.

File layout in bedrock-soc repo:

agents/
  uba/
    __init__.py
    baseline_builder.py    # entry point for job-uba-baseline
    longitudinal_agent.py  # entry point for job-uba-agent
    kql_queries.py         # all KQL query strings, parameterized
    db.py                  # PostgreSQL helpers
    models.py              # dataclasses for Entity, Baseline, Snapshot, Anomaly
    drift.py               # drift scoring logic
    secops_client.py       # SecOps API POST wrapper
    enrichment.py          # optional Azure OpenAI enrichment

baseline_builder.py ​

python
"""
UBA Baseline Builder
Runs weekly as job-uba-baseline in Container Apps.
Queries 30 days of LAW data, computes p50/p95/sigma per metric per entity,
upserts to uba.baselines and uba.daily_snapshots on psql-secops-prod.
"""

import logging
import os
import sys
from datetime import datetime, timezone, timedelta

import numpy as np
import psycopg2
from azure.identity import DefaultAzureCredential
from azure.monitor.query import LogsQueryClient, LogsQueryStatus
from azure.keyvault.secrets import SecretClient

from .kql_queries import QUERIES
from .db import get_connection, upsert_entity, upsert_snapshot, upsert_baseline
from .models import DailySnapshot, Baseline

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
    stream=sys.stdout,
)
log = logging.getLogger("uba.baseline_builder")

# These come from environment variables set in the Container Apps Job definition.
# The job reads them at startup — they're set in Terraform, not hardcoded here.
LAW_WORKSPACE_ID = os.environ["LAW_WORKSPACE_ID"]        # 5d76d1f2
KEY_VAULT_URL    = os.environ["KEY_VAULT_URL"]            # https://cirius-openai-kv-prod.vault.azure.net
DB_SECRET_NAME   = os.environ["DB_SECRET_NAME"]           # uba-db-connection-string
WINDOW_DAYS      = int(os.environ.get("WINDOW_DAYS", "30"))

# Metrics that require special handling: these are ratios (0–1), not raw counts.
# We still store p50/p95/sigma on them, but the drift threshold logic differs.
RATIO_METRICS = {
    "user_offhour_logon_rate",
    "user_failed_logon_rate",
    "user_mfa_challenge_rate",
    "svc_offhour_activity_rate",
}


def get_db_connection_string(credential: DefaultAzureCredential) -> str:
    """Retrieve PostgreSQL connection string from Key Vault at runtime."""
    client = SecretClient(vault_url=KEY_VAULT_URL, credential=credential)
    secret = client.get_secret(DB_SECRET_NAME)
    return secret.value


def query_law(
    logs_client: LogsQueryClient,
    workspace_id: str,
    kql: str,
    start: datetime,
    end: datetime,
) -> list[dict]:
    """Run a KQL query against LAW. Returns list of row dicts. Raises on failure."""
    duration = end - start
    response = logs_client.query_workspace(
        workspace_id=workspace_id,
        query=kql,
        timespan=duration,
    )
    if response.status == LogsQueryStatus.FAILURE:
        raise RuntimeError(f"LAW query failed: {response.partial_error}")

    table = response.tables[0] if response.tables else None
    if table is None:
        return []

    columns = [col.name for col in table.columns]
    return [dict(zip(columns, row)) for row in table.rows]


def compute_baseline(values: list[float]) -> tuple[float, float, float]:
    """
    Compute p50, p95, and sigma from a list of daily values.
    Returns (p50, p95, sigma). Returns (0, 0, 0) if fewer than 7 samples.
    We require at least 7 days before trusting the baseline.
    """
    if len(values) < 7:
        return 0.0, 0.0, 0.0
    arr = np.array(values, dtype=float)
    p50   = float(np.percentile(arr, 50))
    p95   = float(np.percentile(arr, 95))
    sigma = float(np.std(arr, ddof=1)) if len(arr) > 1 else 0.0
    return p50, p95, sigma


def run_baseline_builder() -> None:
    log.info("UBA Baseline Builder starting — window=%d days", WINDOW_DAYS)

    credential  = DefaultAzureCredential()
    logs_client = LogsQueryClient(credential)
    conn_string = get_db_connection_string(credential)
    conn        = get_connection(conn_string)

    end   = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
    start = end - timedelta(days=WINDOW_DAYS)

    log.info("Query window: %s to %s", start.isoformat(), end.isoformat())

    total_entities  = 0
    total_baselines = 0
    errors          = []

    for query_name, query_config in QUERIES.items():
        log.info("Running query: %s", query_name)
        kql = query_config["kql"].replace("START_DATE", start.isoformat()).replace(
            "END_DATE", end.isoformat()
        )
        try:
            rows = query_law(logs_client, LAW_WORKSPACE_ID, kql, start, end)
        except Exception as exc:
            log.error("Query %s failed: %s", query_name, exc)
            errors.append(query_name)
            continue

        log.info("Query %s returned %d rows", query_name, len(rows))

        # Group rows by (entity_type, entity_id, metric)
        # Each row has: entity_id, date, and one or more metric columns
        entity_metric_values: dict[tuple, list[float]] = {}

        for row in rows:
            entity_type = query_config["entity_type"]
            raw_id      = str(row["entity_id"]).strip().upper()

            # Upsert the entity record so we have a DB id to reference
            db_entity_id = upsert_entity(conn, entity_type, raw_id)
            total_entities += 1

            date_val = row["date"]
            if hasattr(date_val, "date"):
                date_val = date_val.date()

            # Each query may produce multiple metric columns per row.
            # Iterate over every metric column defined for this query.
            for metric in query_config["metrics"]:
                raw_value = row.get(metric)
                if raw_value is None:
                    continue
                value = float(raw_value)

                # Store daily snapshot (idempotent upsert)
                snap = DailySnapshot(
                    entity_id=db_entity_id,
                    metric=metric,
                    date=date_val,
                    value=value,
                )
                upsert_snapshot(conn, snap)

                key = (db_entity_id, metric)
                entity_metric_values.setdefault(key, []).append(value)

        # Compute and upsert baselines for all entity/metric pairs from this query
        for (db_entity_id, metric), values in entity_metric_values.items():
            p50, p95, sigma = compute_baseline(values)
            bl = Baseline(
                entity_id    = db_entity_id,
                metric       = metric,
                window_days  = WINDOW_DAYS,
                p50          = p50,
                p95          = p95,
                sigma        = sigma,
                sample_count = len(values),
            )
            upsert_baseline(conn, bl)
            total_baselines += 1

    conn.commit()
    conn.close()

    log.info(
        "Baseline builder complete — entities=%d, baselines=%d, errors=%d",
        total_entities,
        total_baselines,
        len(errors),
    )
    if errors:
        log.warning("Failed queries (will retry next week): %s", errors)
        # Do not exit 1 — partial baselines are better than none.
        # Alerting on job failure is handled by the Container Apps Job execution history.


if __name__ == "__main__":
    run_baseline_builder()

db.py ​

python
"""
PostgreSQL helpers for the UBA system.
All writes are explicit — no ORM. Keeps the schema surface visible.
"""

import logging
from datetime import date
from typing import Optional

import psycopg2
import psycopg2.extras

from .models import DailySnapshot, Baseline

log = logging.getLogger("uba.db")


def get_connection(connection_string: str):
    """Return a psycopg2 connection. Caller owns the lifecycle."""
    conn = psycopg2.connect(connection_string)
    conn.autocommit = False
    return conn


def upsert_entity(conn, entity_type: str, entity_id: str) -> int:
    """
    Ensure an entity row exists. Returns the integer primary key.
    Updates last_activity on conflict.
    """
    sql = """
        INSERT INTO uba.entities (entity_type, entity_id)
        VALUES (%s, %s)
        ON CONFLICT (entity_type, entity_id)
        DO UPDATE SET last_activity = NOW()
        RETURNING id
    """
    with conn.cursor() as cur:
        cur.execute(sql, (entity_type, entity_id))
        return cur.fetchone()[0]


def upsert_snapshot(conn, snap: DailySnapshot) -> None:
    """Upsert a daily snapshot. Idempotent on (entity_id, metric, date)."""
    sql = """
        INSERT INTO uba.daily_snapshots (entity_id, metric, date, value)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT (entity_id, metric, date)
        DO UPDATE SET value = EXCLUDED.value
    """
    with conn.cursor() as cur:
        cur.execute(sql, (snap.entity_id, snap.metric, snap.date, snap.value))


def upsert_baseline(conn, bl: Baseline) -> None:
    """Upsert a baseline. On conflict (entity_id, metric), update all fields."""
    sql = """
        INSERT INTO uba.baselines
            (entity_id, metric, window_days, p50, p95, sigma, sample_count, computed_at)
        VALUES (%s, %s, %s, %s, %s, %s, %s, NOW())
        ON CONFLICT (entity_id, metric)
        DO UPDATE SET
            window_days  = EXCLUDED.window_days,
            p50          = EXCLUDED.p50,
            p95          = EXCLUDED.p95,
            sigma        = EXCLUDED.sigma,
            sample_count = EXCLUDED.sample_count,
            computed_at  = NOW()
        WHERE uba.baselines.suppressed = FALSE
           OR uba.baselines.suppress_until < NOW()
    """
    with conn.cursor() as cur:
        cur.execute(
            sql,
            (bl.entity_id, bl.metric, bl.window_days, bl.p50, bl.p95, bl.sigma, bl.sample_count),
        )


def get_baselines_for_entities(
    conn, entity_type: str
) -> dict[tuple[str, str], dict]:
    """
    Return all non-suppressed baselines for an entity type.
    Returns dict keyed by (entity_id_str, metric) -> baseline dict.
    """
    sql = """
        SELECT e.entity_id, b.metric, b.p50, b.p95, b.sigma, b.sample_count
        FROM uba.baselines b
        JOIN uba.entities e ON e.id = b.entity_id
        WHERE e.entity_type = %s
          AND b.suppressed = FALSE
          AND (b.suppress_until IS NULL OR b.suppress_until < NOW())
          AND b.sample_count >= 7
    """
    with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
        cur.execute(sql, (entity_type,))
        rows = cur.fetchall()

    result = {}
    for row in rows:
        key = (row["entity_id"], row["metric"])
        result[key] = dict(row)
    return result


def insert_anomaly(conn, anomaly: dict) -> int:
    """Insert a new anomaly record. Returns the new anomaly id."""
    sql = """
        INSERT INTO uba.anomalies
            (entity_id, metric, analysis_date, severity, observed_value,
             baseline_p50, baseline_p95, baseline_sigma, z_score, ioc_key, explanation)
        VALUES (%(entity_id)s, %(metric)s, %(analysis_date)s, %(severity)s,
                %(observed_value)s, %(baseline_p50)s, %(baseline_p95)s,
                %(baseline_sigma)s, %(z_score)s, %(ioc_key)s, %(explanation)s)
        ON CONFLICT (ioc_key) DO NOTHING
        RETURNING id
    """
    with conn.cursor() as cur:
        cur.execute(sql, anomaly)
        row = cur.fetchone()
        return row[0] if row else -1


def mark_anomaly_posted(conn, anomaly_db_id: int, incident_id: int) -> None:
    """Record that an anomaly was successfully posted to SecOps."""
    sql = """
        UPDATE uba.anomalies
        SET incident_id = %s, posted_at = NOW()
        WHERE id = %s
    """
    with conn.cursor() as cur:
        cur.execute(sql, (incident_id, anomaly_db_id))

Longitudinal UBA Agent — Python Implementation ​

The longitudinal agent runs daily inside the existing Container Apps Job schedule. It queries the last 24 hours from LAW, compares against baselines, and posts anomalies.

longitudinal_agent.py ​

python
"""
UBA Longitudinal Analysis Agent
Runs daily in ca-secops-prod Container Apps Job.
Queries last 24h from LAW, computes drift scores vs. baselines,
posts anomalies to soc.bedrockcybersecurity.org/api/incidents.
"""

import logging
import os
import sys
from datetime import datetime, timezone, timedelta, date

from azure.identity import DefaultAzureCredential
from azure.monitor.query import LogsQueryClient
from azure.keyvault.secrets import SecretClient

from .kql_queries import QUERIES
from .db import (
    get_connection,
    upsert_entity,
    upsert_snapshot,
    get_baselines_for_entities,
    insert_anomaly,
    mark_anomaly_posted,
)
from .drift import compute_drift_severity, compute_z_score
from .secops_client import post_incident
from .enrichment import enrich_anomaly  # no-op if enrichment disabled

log = logging.getLogger("uba.longitudinal_agent")

LAW_WORKSPACE_ID = os.environ["LAW_WORKSPACE_ID"]
KEY_VAULT_URL    = os.environ["KEY_VAULT_URL"]
DB_SECRET_NAME   = os.environ["DB_SECRET_NAME"]
SECOPS_API_URL   = os.environ.get("SECOPS_API_URL", "https://secops.bedrockcybersecurity.org")
SECOPS_API_KEY   = os.environ.get("SECOPS_API_KEY", "")  # set via Key Vault reference
ENRICHMENT       = os.environ.get("UBA_ENRICHMENT", "false").lower() == "true"
SOURCE_SYSTEM    = os.environ.get("SOURCE_SYSTEM", "PROD")


def get_db_connection_string(credential):
    client = SecretClient(vault_url=KEY_VAULT_URL, credential=credential)
    return client.get_secret(DB_SECRET_NAME).value


def run_longitudinal_agent() -> None:
    log.info("UBA Longitudinal Agent starting")

    credential  = DefaultAzureCredential()
    logs_client = LogsQueryClient(credential)
    conn_string = get_db_connection_string(credential)
    conn        = get_connection(conn_string)

    # Analysis window: yesterday (full 24h UTC day)
    today       = datetime.now(timezone.utc).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    start       = today - timedelta(days=1)
    end         = today
    analysis_date = start.date()

    log.info("Analysis window: %s to %s", start.isoformat(), end.isoformat())

    anomalies_detected = 0
    anomalies_posted   = 0

    for entity_type in ("user", "device", "service_account"):
        # Load baselines for this entity type from PostgreSQL
        baselines = get_baselines_for_entities(conn, entity_type)
        if not baselines:
            log.info("No baselines found for entity_type=%s — skipping", entity_type)
            continue

        # Query current 24h from LAW for all queries covering this entity type
        current_values: dict[tuple[str, str], float] = {}

        for query_name, query_config in QUERIES.items():
            if query_config["entity_type"] != entity_type:
                continue

            kql = query_config["kql"].replace(
                "START_DATE", start.isoformat()
            ).replace("END_DATE", end.isoformat())

            try:
                from .baseline_builder import query_law
                rows = query_law(logs_client, LAW_WORKSPACE_ID, kql, start, end)
            except Exception as exc:
                log.error("Query %s failed: %s", query_name, exc)
                continue

            for row in rows:
                raw_id   = str(row["entity_id"]).strip().upper()
                db_id    = upsert_entity(conn, entity_type, raw_id)
                date_val = analysis_date

                for metric in query_config["metrics"]:
                    raw_value = row.get(metric)
                    if raw_value is None:
                        continue
                    value = float(raw_value)
                    upsert_snapshot(conn, type("S", (), {
                        "entity_id": db_id,
                        "metric": metric,
                        "date": date_val,
                        "value": value,
                    })())
                    current_values[(raw_id, metric)] = value

        # Compare each current value against its baseline
        for (entity_id_str, metric), observed in current_values.items():
            bl = baselines.get((entity_id_str, metric))
            if bl is None:
                # No baseline yet — this entity is new. Skip detection.
                continue

            severity = compute_drift_severity(
                metric=metric,
                observed=observed,
                p95=bl["p95"],
                sigma=bl["sigma"],
                entity_type=entity_type,
            )

            if severity is None:
                continue

            z_score = compute_z_score(observed, bl["p95"], bl["sigma"])
            ioc_key = (
                f"UBA:{entity_type}:{entity_id_str}:{metric}:{analysis_date.isoformat()}"
            )

            # Optional LLM enrichment for HIGH/CRITICAL
            explanation = None
            if ENRICHMENT and severity in ("HIGH", "CRITICAL"):
                explanation = enrich_anomaly(
                    entity_type=entity_type,
                    entity_id=entity_id_str,
                    metric=metric,
                    observed=observed,
                    baseline=bl,
                    severity=severity,
                )

            db_entity_id = upsert_entity(conn, entity_type, entity_id_str)
            anomaly_record = {
                "entity_id":      db_entity_id,
                "metric":         metric,
                "analysis_date":  analysis_date,
                "severity":       severity,
                "observed_value": observed,
                "baseline_p50":   bl["p50"],
                "baseline_p95":   bl["p95"],
                "baseline_sigma": bl["sigma"],
                "z_score":        z_score,
                "ioc_key":        ioc_key,
                "explanation":    explanation,
            }

            anomaly_db_id = insert_anomaly(conn, anomaly_record)
            if anomaly_db_id == -1:
                log.debug("Anomaly already exists for ioc_key=%s — skipping post", ioc_key)
                continue

            anomalies_detected += 1

            # Post to SecOps
            incident_payload = build_incident_payload(
                entity_type=entity_type,
                entity_id=entity_id_str,
                metric=metric,
                severity=severity,
                observed=observed,
                baseline=bl,
                z_score=z_score,
                ioc_key=ioc_key,
                analysis_date=analysis_date,
                explanation=explanation,
            )

            try:
                incident_id = post_incident(SECOPS_API_URL, SECOPS_API_KEY, incident_payload)
                mark_anomaly_posted(conn, anomaly_db_id, incident_id)
                anomalies_posted += 1
                log.info("Posted incident %d for %s — %s", incident_id, entity_id_str, metric)
            except Exception as exc:
                log.error("Failed to post incident for ioc_key=%s: %s", ioc_key, exc)
                # Continue processing — don't let one post failure stop the run

    conn.commit()
    conn.close()

    log.info(
        "Longitudinal agent complete — detected=%d, posted=%d",
        anomalies_detected,
        anomalies_posted,
    )


def build_incident_payload(
    entity_type: str,
    entity_id: str,
    metric: str,
    severity: str,
    observed: float,
    baseline: dict,
    z_score: float,
    ioc_key: str,
    analysis_date: date,
    explanation: str | None,
) -> dict:
    """Build the SecOps /api/incidents payload."""
    metric_label = metric.replace("_", " ").title()
    description_lines = [
        f"**UBA Behavioral Drift Detected**",
        f"",
        f"- **Entity:** {entity_id} ({entity_type})",
        f"- **Metric:** {metric_label}",
        f"- **Analysis Date:** {analysis_date.isoformat()}",
        f"- **Observed Value:** {observed:.2f}",
        f"- **Baseline p50:** {baseline['p50']:.2f}",
        f"- **Baseline p95:** {baseline['p95']:.2f}",
        f"- **Sigma:** {baseline['sigma']:.2f}",
        f"- **Z-Score above p95:** {z_score:.2f}σ",
        f"- **Sample Count (baseline):** {baseline['sample_count']} days",
    ]
    if explanation:
        description_lines += ["", "**AI Analysis:**", explanation]

    return {
        "title":         f"UBA: {metric_label} drift — {entity_id}",
        "description":   "\n".join(description_lines),
        "severity":      severity,
        "source_system": SOURCE_SYSTEM,
        "ioc_key":       ioc_key,
        "entity_type":   entity_type,
        "entity_id":     entity_id,
        "tags":          ["UBA", "behavioral-analytics", entity_type],
    }


if __name__ == "__main__":
    run_longitudinal_agent()

drift.py ​

python
"""
Drift scoring logic.
Kept in its own module so thresholds can be adjusted without touching agent code.
"""

from typing import Optional

# Metrics that are always CRITICAL regardless of baseline
ALWAYS_CRITICAL = {
    "svc_interactive_logon_daily",  # service accounts never have interactive logons
}

# Metrics where any non-zero value on a new-country event is HIGH.
# These are injected by a separate KQL check, not computed from baseline.
ALWAYS_HIGH_NEW_COUNTRY = True  # flag controlled at call site


def compute_drift_severity(
    metric: str,
    observed: float,
    p95: float,
    sigma: float,
    entity_type: str,
    new_country: bool = False,
) -> Optional[str]:
    """
    Compute severity for a single observed value vs. its baseline.
    Returns: 'CRITICAL' | 'HIGH' | 'MEDIUM' | 'LOW' | None (no anomaly)

    Thresholds:
        observed > p95 + 2σ  → HIGH
        observed > p95 + 1σ  → MEDIUM
        observed > p95       → LOW
        otherwise            → None

    Special cases that override thresholds:
        svc_interactive_logon_daily any value > 0 → CRITICAL
        new_country=True → HIGH minimum
    """
    if metric in ALWAYS_CRITICAL:
        if observed > 0:
            return "CRITICAL"
        return None

    if new_country:
        # Activity from a new country is always at least HIGH
        return "HIGH"

    # Guard: if sigma is tiny (< 0.01), treat as zero to avoid division issues
    effective_sigma = max(sigma, 0.01)

    if observed > p95 + 2 * effective_sigma:
        return "HIGH"
    elif observed > p95 + 1 * effective_sigma:
        return "MEDIUM"
    elif observed > p95:
        return "LOW"
    return None


def compute_z_score(observed: float, p95: float, sigma: float) -> float:
    """How many standard deviations above p95 is the observed value?"""
    effective_sigma = max(sigma, 0.01)
    return (observed - p95) / effective_sigma

Severity Mapping ​

ConditionSeverityNotes
svc_interactive_logon_daily > 0CRITICALAlways, no threshold
Activity from new country/regionHIGHAlways, checked separately
observed > p95 + 2σHIGHStandard threshold
observed > p95 + 1σMEDIUMStandard threshold
observed > p95LOWMonitor, may tune to suppress

Severity maps directly to the SecOps incident severity field. CRITICAL and HIGH incidents wake the Telegram notification channel. LOW incidents are visible in SecOps but do not trigger notifications by default — adjust in SecOps notification rules.

ioc_key Format ​

UBA:{entity_type}:{entity_id}:{metric}:{date}

Examples:

UBA:user:RORY@CIRIUSGROUP.COM:user_logon_count_daily:2026-05-25
UBA:device:WORKSTATION-42:device_bytes_sent_daily:2026-05-25
UBA:service_account:CIRIUSGROUP\SVC-BACKUP:svc_interactive_logon_daily:2026-05-25

The ioc_key is unique per entity/metric/day. If the agent is rerun on the same day (e.g., after a failure), insert_anomaly uses ON CONFLICT (ioc_key) DO NOTHING, so the incident is posted only once. The mark_anomaly_posted update runs separately only when the SecOps POST succeeds.

The source_system field on the incident payload is separate from the ioc_key. Include source_system: PROD (or DDE/AWS if the UBA agent runs in other environments) — the SecurityAgent uses it to route suppression checks correctly.

Deployment ​

Prerequisites ​

Before deploying either job:

  1. uba PostgreSQL schema is created (run the DDL in Database Schema)
  2. A PostgreSQL connection string secret exists in Key Vault (cirius-openai-kv-prod):
    • Secret name: uba-db-connection-string
    • Value: host=psql-secops-prod.postgres.database.azure.com dbname=secops user=uba_agent password=... sslmode=require
  3. A SecOps API key secret exists in Key Vault:
    • Secret name: secops-api-key (already exists if other agents are running)
  4. The managed identity for the job has Monitoring Reader on cirius-logging-law-central
  5. The managed identity has Key Vault Secrets User on cirius-openai-kv-prod

Container Image ​

Both jobs use the same container image as ca-secops-prod. The entry point changes per job. In the Container Apps Job definition (Terraform), set the command override:

hcl
# For job-uba-baseline:
command = ["python", "-m", "agents.uba.baseline_builder"]

# For job-uba-agent:
command = ["python", "-m", "agents.uba.longitudinal_agent"]

Terraform — job-uba-baseline ​

Add to the appropriate Terraform module in azure-infra. Pattern matches existing Container Apps Job definitions (see job-orchestrator-cirius).

hcl
resource "azurerm_container_app_job" "uba_baseline" {
  name                = "job-uba-baseline"
  resource_group_name = "rg-logging-logs"
  location            = var.location
  container_app_environment_id = azurerm_container_app_environment.main.id

  replica_timeout_in_seconds = 3600  # 1 hour max
  replica_retry_limit        = 1

  schedule_trigger_config {
    cron_expression          = "0 2 * * 0"  # Every Sunday at 02:00 UTC
    parallelism              = 1
    replica_completion_count = 1
  }

  identity {
    type = "UserAssigned"
    identity_ids = [azurerm_user_assigned_identity.secops_jobs.id]
  }

  template {
    container {
      name   = "uba-baseline"
      image  = "${var.acr_login_server}/soc-web:latest"
      cpu    = 1.0
      memory = "2Gi"
      command = ["python", "-m", "agents.uba.baseline_builder"]

      env {
        name  = "LAW_WORKSPACE_ID"
        value = "5d76d1f2"
      }
      env {
        name  = "KEY_VAULT_URL"
        value = "https://cirius-openai-kv-prod.vault.azure.net"
      }
      env {
        name  = "DB_SECRET_NAME"
        value = "uba-db-connection-string"
      }
      env {
        name  = "WINDOW_DAYS"
        value = "30"
      }
      env {
        name        = "SOURCE_SYSTEM"
        value       = "PROD"
      }
    }
  }

  tags = local.common_tags
}

Add a matching job-uba-agent resource with:

  • cron_expression = "0 6 * * *" (daily at 06:00 UTC — after LAW ingestion completes)
  • command = ["python", "-m", "agents.uba.longitudinal_agent"]
  • Same environment variables plus SECOPS_API_URL and a Key Vault reference for SECOPS_API_KEY

Key Vault References for Secrets ​

For secrets that must not appear in environment variables as plaintext, use the Container Apps secrets-as-key-vault-references pattern. The SECOPS_API_KEY is a good candidate:

hcl
secret {
  name                = "secops-api-key"
  key_vault_secret_id = "${azurerm_key_vault.main.vault_uri}secrets/secops-api-key"
  identity            = azurerm_user_assigned_identity.secops_jobs.id
}

Then reference it as:

hcl
env {
  name        = "SECOPS_API_KEY"
  secret_name = "secops-api-key"
}

Verifying Deployment ​

After Terraform apply:

bash
# Verify job-uba-baseline exists
az containerapp job show \
  --name job-uba-baseline \
  --resource-group rg-logging-logs \
  --output table

# Manually trigger first baseline run (don't wait for Sunday)
az containerapp job start \
  --name job-uba-baseline \
  --resource-group rg-logging-logs

# Watch logs
az monitor log-analytics query \
  --workspace 5d76d1f2 \
  --analytics-query "ContainerAppConsoleLogs_CL | where ContainerAppName_s == 'job-uba-baseline' | sort by TimeGenerated asc" \
  --output table

# After baseline run: verify rows in PostgreSQL
# psql into psql-secops-prod:
SELECT entity_type, COUNT(*) as entities, COUNT(DISTINCT metric) as metrics
FROM uba.entities e
JOIN uba.baselines b ON b.entity_id = e.id
GROUP BY entity_type;

Bootstrapping — Initial 30-Day Baseline ​

When deploying for the first time, there is no baseline data. The baseline builder queries 30 days of LAW history on its first run — this is historical data that is already there, not data you need to generate. The process is:

  1. Deploy the schema (run the DDL)
  2. Trigger job-uba-baseline manually
  3. The job queries cirius-logging-law-central for the last 30 days of SignInLogs, SecurityEvent, OfficeActivity, and CommonSecurityLog
  4. After the job completes (~20–40 minutes depending on tenant size), check that uba.baselines has rows with sample_count >= 7
  5. Entities with fewer than 7 samples are marked with baseline_valid = false and will be skipped by the longitudinal agent until the next baseline run catches up

Entities that won't have a 30-day baseline on day one:

  • New employees hired in the last 30 days
  • Devices enrolled in the last 30 days
  • Service accounts created in the last 30 days

These entities accumulate daily snapshots from the longitudinal agent but are not scored for drift until sample_count >= 7. After the next weekly baseline run, they will have enough data to score.

If LAW doesn't have 30 days of data: This happens if you just migrated to LAW or changed your data collection configuration. In this case, run the baseline builder weekly for 4 weeks before enabling the longitudinal agent. The longitudinal agent will produce no anomalies (no baselines with sufficient samples), which is the right behavior — better no alerts than a flood of false positives from an undertrained baseline.

Backfilling Historical Snapshots ​

If you want to accelerate baseline training by loading historical data, run the baseline builder with an extended window:

bash
# Override WINDOW_DAYS for the first run only — 90 days of history
az containerapp job start \
  --name job-uba-baseline \
  --resource-group rg-logging-logs \
  --environment-variables "WINDOW_DAYS=90"

Verify LAW retention covers 90 days before attempting this. Default LAW retention is 90 days; check Portal → cirius-logging-law-central → Usage and estimated costs → Data retention.

Tuning ​

Adjusting Thresholds ​

Thresholds are defined in drift.py. To adjust:

  1. Edit compute_drift_severity in agents/uba/drift.py
  2. Test locally with pytest agents/uba/tests/test_drift.py
  3. PR to bedrock-soc, merge, redeploy

The threshold parameters to tune:

  • 2 * effective_sigma → HIGH boundary. Raise to reduce HIGH alerts.
  • 1 * effective_sigma → MEDIUM boundary. Raise to reduce MEDIUM alerts.
  • LOW threshold (currently any value > p95) → consider removing LOW entirely if noise is too high. Start with LOW enabled for two weeks, then evaluate.

Suppressing a Metric for an Entity ​

When an entity has a known legitimate reason to exceed its baseline (role change, project deadline, vacation return), suppress the metric rather than raising a false positive:

sql
-- Suppress for 14 days
UPDATE uba.baselines
SET suppressed = TRUE,
    suppress_reason = 'Role change to Project Lead — elevated Teams/SharePoint activity expected',
    suppress_until = NOW() + INTERVAL '14 days'
WHERE entity_id = (SELECT id FROM uba.entities WHERE entity_id = 'RORY@CIRIUSGROUP.COM')
  AND metric = 'user_m365_sharepoint_events_daily';

The baseline builder will not overwrite suppressed = TRUE rows until suppress_until has passed (see the WHERE clause in upsert_baseline).

Handling New Employees (No Baseline) ​

New employees are detected automatically: the longitudinal agent calls upsert_entity, which creates the entity row, then get_baselines_for_entities returns nothing for them (no baselines yet), so they are skipped. No special configuration needed.

During the first 7 days: no scoring. After the first baseline rebuild (next Sunday): scoring begins with whatever sample count has accumulated. If they joined mid-week, they may have 5–6 days of samples and still be skipped until the following Sunday.

To force a new employee into scoring faster after 14+ days:

bash
# Trigger baseline rebuild early
az containerapp job start \
  --name job-uba-baseline \
  --resource-group rg-logging-logs

Handling Authorized Out-of-Pattern Activity ​

Vacation return: Suppress user_logon_count_daily and user_offhour_logon_rate for 3–5 days after a long absence. The return surge will otherwise fire as HIGH.

Role changes (Entra/PIM): Suppress user_pim_activations_daily for the transition period. Set suppress_until to 30 days after the role change date.

Planned data migration: Suppress device_bytes_sent_daily for the device(s) running the migration. Set suppress_until to the expected completion date.

Service account source host expansion: If a service account is now authorized to run from additional hosts (e.g., a new application server), update the baseline manually after the change:

sql
-- Re-enable baseline computation after legitimate source expansion
UPDATE uba.baselines
SET suppressed = FALSE, suppress_reason = NULL, suppress_until = NULL
WHERE entity_id = (
    SELECT id FROM uba.entities
    WHERE entity_id = 'CIRIUSGROUP\\SVC-BACKUP'
      AND entity_type = 'service_account'
)
AND metric = 'svc_unique_src_hosts_daily';
-- Then trigger a baseline rebuild to incorporate the new source

Integration with SecOps ​

Incident Payload ​

The longitudinal agent POSTs to https://soc.bedrockcybersecurity.org/api/incidents using the X-API-Key header. The payload format matches what other agents send:

json
{
  "title": "UBA: User Logon Count Daily drift — RORY@CIRIUSGROUP.COM",
  "description": "**UBA Behavioral Drift Detected**\n\n- **Entity:** RORY@CIRIUSGROUP.COM (user)\n- **Metric:** User Logon Count Daily\n- **Analysis Date:** 2026-05-25\n- **Observed Value:** 47.00\n- **Baseline p50:** 8.00\n- **Baseline p95:** 18.00\n- **Sigma:** 3.20\n- **Z-Score above p95:** 9.06σ\n- **Sample Count (baseline):** 30 days",
  "severity": "HIGH",
  "source_system": "PROD",
  "ioc_key": "UBA:user:RORY@CIRIUSGROUP.COM:user_logon_count_daily:2026-05-25",
  "entity_type": "user",
  "entity_id": "RORY@CIRIUSGROUP.COM",
  "tags": ["UBA", "behavioral-analytics", "user"]
}

Known-Good Rules ​

UBA anomalies can be suppressed by SecOps known-good rules just like any other incident source. The SecurityAgent checks GET /api/known-good on startup.

To suppress a recurring false positive at the SecOps level (not just the DB level): SecOps UI → Known Good Rules → Add Rule → ioc_key prefix: UBA:user:{upn}:{metric}

This suppresses all future detections for that entity/metric pair without touching the UBA database.

False Positive Feedback Loop ​

When an analyst marks an incident as a false positive in SecOps, the SecOps platform should propagate that signal back to UBA. Implement this as a webhook or daily batch job that queries SecOps for newly closed incidents with source UBA and updates uba.anomalies.false_positive = true. This data feeds future threshold tuning.

Until the webhook is built, mark false positives manually:

sql
UPDATE uba.anomalies
SET false_positive = TRUE,
    fp_reason = 'Analyst confirmed — vacation return',
    fp_marked_at = NOW(),
    fp_marked_by = 'rory@ciriusgroup.com'
WHERE ioc_key = 'UBA:user:RORY@CIRIUSGROUP.COM:user_logon_count_daily:2026-05-25';

HIPAA Considerations ​

UBA processes behavioral metadata derived from PHI-adjacent systems: sign-in logs for users who access healthcare data, device activity for machines that touch ePHI, service account behavior for accounts that access clinical systems.

What to Log (Safe) ​

  • Timestamps, event counts, aggregates — all fine. These are operational metrics, not content.
  • Entity identifiers (UPNs, hostnames, SAM accounts) — fine. Already in your SIEM.
  • Behavioral statistics (p50, p95, sigma) — fine. No PHI content.
  • Anomaly records (observed value, severity, ioc_key) — fine.

What NOT to Log ​

  • Never log the content of M365 activity. OfficeActivity gives you event counts and app usage patterns. Do not log filenames, document titles, email subjects, or message content. The KQL queries above aggregate counts only — keep it that way.
  • Never log process command lines in the UBA database. EventID 4688 command line arguments can contain file paths, usernames, or content passed as arguments. Log process counts only.
  • Never log raw LAW query results to stdout in production. The baseline builder and longitudinal agent log aggregate statistics, not row-level data. Verify this before enabling debug logging.

Retention ​

DataRetentionBasis
uba.daily_snapshots90 daysOperational — needed for baseline window
uba.baselinesPermanentStatistical models, not event data
uba.anomalies6 yearsHIPAA audit trail requirement
uba.entitiesPermanentEntity registry, no PHI

Prune uba.daily_snapshots older than 90 days weekly:

sql
DELETE FROM uba.daily_snapshots
WHERE date < CURRENT_DATE - INTERVAL '90 days';

Add this as a scheduled query in the baseline builder (run at end of job) or as a PostgreSQL cron extension job.

Minimum Necessary Access ​

The managed identity for UBA jobs should have read-only access to LAW and write access to only the uba schema in PostgreSQL. It should not have access to the public schema or any SecOps platform tables. Create a dedicated PostgreSQL role:

sql
-- Run once on psql-secops-prod
CREATE ROLE uba_agent LOGIN;
GRANT CONNECT ON DATABASE secops TO uba_agent;
GRANT USAGE ON SCHEMA uba TO uba_agent;
GRANT SELECT, INSERT, UPDATE ON ALL TABLES IN SCHEMA uba TO uba_agent;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA uba TO uba_agent;
-- Revoke access to public schema (secops platform tables)
REVOKE ALL ON SCHEMA public FROM uba_agent;

Store the uba_agent credentials as the uba-db-connection-string secret in Key Vault — not the superuser or the secops_app user credentials.

BAA and Vendor Scope ​

Azure Log Analytics, Azure Database for PostgreSQL, and Azure OpenAI are all covered under Microsoft's BAA for Azure. If the optional LLM enrichment is enabled, anomaly descriptions (not raw content) are sent to Azure OpenAI. The enrichment prompt should describe behavioral statistics only — never include M365 content, process arguments, or anything that could constitute ePHI.

Monitoring and Alerting ​

Job Execution Monitoring ​

Monitor both jobs the same way as other Container Apps Jobs. See Container Apps Jobs Monitoring for the full procedure. Key checks:

bash
# Check baseline job ran this week
az containerapp job execution list \
  --name job-uba-baseline \
  --resource-group rg-logging-logs \
  --output table \
  --query "sort_by([].{Name:name,StartTime:properties.startTime,Status:properties.status}, &StartTime) | [-5:]"

# Check daily agent ran today
az containerapp job execution list \
  --name job-uba-agent \
  --resource-group rg-logging-logs \
  --output table \
  --query "sort_by([].{Name:name,StartTime:properties.startTime,Status:properties.status}, &StartTime) | [-3:]"

Baseline Health Check Query ​

Run weekly to verify baselines are being maintained:

sql
SELECT
    entity_type,
    COUNT(*) as entity_count,
    AVG(sample_count) as avg_sample_count,
    MIN(computed_at) as oldest_baseline,
    MAX(computed_at) as newest_baseline,
    COUNT(*) FILTER (WHERE sample_count < 7) as insufficient_samples,
    COUNT(*) FILTER (WHERE suppressed) as suppressed_count
FROM uba.baselines b
JOIN uba.entities e ON e.id = b.entity_id
GROUP BY entity_type
ORDER BY entity_type;

If oldest_baseline is more than 8 days ago, the baseline builder failed or missed a run.

LAW Query for Anomaly Throughput ​

After the longitudinal agent runs, verify it's finding and posting anomalies:

bash
az monitor log-analytics query \
  --workspace 5d76d1f2 \
  --analytics-query "ContainerAppConsoleLogs_CL | where ContainerAppName_s == 'job-uba-agent' | where Log_s contains 'Posted incident' | where TimeGenerated > ago(26h) | count" \
  --output table

A count of zero after a full run is unusual unless the environment is very quiet. Check uba.anomalies directly:

sql
SELECT severity, COUNT(*) as count
FROM uba.anomalies
WHERE analysis_date = CURRENT_DATE - 1
GROUP BY severity
ORDER BY severity;

Common Failure Patterns ​

Baseline Builder Returns Zero Rows ​

Symptom: Job completes successfully but uba.baselines has no new rows.

Cause: KQL queries returned no data. Usually a LAW ingestion gap, or the time window is wrong (off-by-one on UTC date math).

Fix:

  1. Run the query manually in LAW portal: Log Analytics → cirius-logging-law-central → Logs — paste the KQL with explicit dates
  2. Verify data exists in SignInLogs/SecurityEvent for the window
  3. Check LAW ingestion health: Portal → cirius-logging-law-central → Usage

Longitudinal Agent Posts No Incidents ​

Symptom: Agent runs, no errors, zero anomalies detected.

Causes:

  • No baselines with sample_count >= 7 — first week, expected behavior
  • All entities are within their p95 — possibly a quiet day, or threshold needs tuning
  • LAW query returned no 24h data — check ingestion for yesterday

Fix: Query uba.daily_snapshots for yesterday's date. If empty, the LAW query failed silently. Check agent logs for Query ... returned 0 rows warnings.

SecOps POST Fails ​

Symptom: Anomalies appear in uba.anomalies with incident_id = null and posted_at = null.

Fix:

  1. Check SecOps platform health: curl -I https://secops.bedrockcybersecurity.org
  2. Verify the API key secret is current in Key Vault
  3. Manually repost unposted anomalies: query uba.anomalies WHERE posted_at IS NULL and post each via curl -X POST https://soc.bedrockcybersecurity.org/api/incidents

Service Account Interactive Logon False Positive ​

Symptom: svc_interactive_logon_daily CRITICAL fires for a service account that legitimately logs in interactively during maintenance.

This should not happen — service accounts should never have interactive logons. Investigate before suppressing. If it's genuinely authorized:

sql
-- Suppress for a specific maintenance window only
UPDATE uba.baselines
SET suppressed = TRUE,
    suppress_reason = 'Authorized maintenance window — DB migration 2026-05-28',
    suppress_until = '2026-05-29 06:00:00+00'
WHERE entity_id = (
    SELECT id FROM uba.entities
    WHERE entity_id = 'CIRIUSGROUP\\SVC-DBMAINT'
      AND entity_type = 'service_account'
)
AND metric = 'svc_interactive_logon_daily';

If the account routinely has interactive logons, it is not a service account — review the account type and alert the identity team.

Document History ​

DateChangeAuthor
May 2026Initial draft — database schema, baseline builder implementation, KQL queries for all metric categories, longitudinal agent implementation, severity mapping, ioc_key format, deployment (Terraform), bootstrapping, tuning, SecOps integration, HIPAA considerations. System is planned, not yet deployed.Rory

Internal use only — Cirius Group