Add Censys Supplier Risk solution

.script/tests/KqlvalidationsTests/CustomTables/CensysRiskEvents_CL.json

@@ -0,0 +1,149 @@
+{
+  "Name": "CensysRiskEvents_CL",
+  "Properties": [
+    {
+      "Name": "TimeGenerated",
+      "Type": "datetime"
+    },
+    {
+      "Name": "EventId",
+      "Type": "string"
+    },
+    {
+      "Name": "InsertedAt",
+      "Type": "datetime"
+    },
+    {
+      "Name": "OccurredAt",
+      "Type": "datetime"
+    },
+    {
+      "Name": "ChangeType",
+      "Type": "string"
+    },
+    {
+      "Name": "RiskInstanceId",
+      "Type": "string"
+    },
+    {
+      "Name": "SupplyChainId",
+      "Type": "string"
+    },
+    {
+      "Name": "AssetType",
+      "Type": "string"
+    },
+    {
+      "Name": "RiskTypeId",
+      "Type": "string"
+    },
+    {
+      "Name": "CveId",
+      "Type": "string"
+    },
+    {
+      "Name": "RiskCategory",
+      "Type": "string"
+    },
+    {
+      "Name": "RiskName",
+      "Type": "string"
+    },
+    {
+      "Name": "Description",
+      "Type": "string"
+    },
+    {
+      "Name": "Remediations",
+      "Type": "dynamic"
+    },
+    {
+      "Name": "Severity",
+      "Type": "string"
+    },
+    {
+      "Name": "Score",
+      "Type": "real"
+    },
+    {
+      "Name": "Confidence",
+      "Type": "real"
+    },
+    {
+      "Name": "Status",
+      "Type": "string"
+    },
+    {
+      "Name": "ResolveTime",
+      "Type": "datetime"
+    },
+    {
+      "Name": "ResolveReason",
+      "Type": "string"
+    },
+    {
+      "Name": "SupplierId",
+      "Type": "string"
+    },
+    {
+      "Name": "SupplierDisplayName",
+      "Type": "string"
+    },
+    {
+      "Name": "Suppliers",
+      "Type": "dynamic"
+    },
+    {
+      "Name": "AssetIp",
+      "Type": "string"
+    },
+    {
+      "Name": "AssetHostname",
+      "Type": "string"
+    },
+    {
+      "Name": "AssetPort",
+      "Type": "int"
+    },
+    {
+      "Name": "ServiceTransportProtocol",
+      "Type": "string"
+    },
+    {
+      "Name": "ServicePort",
+      "Type": "int"
+    },
+    {
+      "Name": "ServiceProtocol",
+      "Type": "string"
+    },
+    {
+      "Name": "Tags",
+      "Type": "dynamic"
+    },
+    {
+      "Name": "RapidResponse",
+      "Type": "bool"
+    },
+    {
+      "Name": "Evidence",
+      "Type": "string"
+    },
+    {
+      "Name": "FirstDetectedAt",
+      "Type": "datetime"
+    },
+    {
+      "Name": "LastDetectedAt",
+      "Type": "datetime"
+    },
+    {
+      "Name": "CreatedAt",
+      "Type": "datetime"
+    },
+    {
+      "Name": "PlatformUrl",
+      "Type": "string"
+    }
+  ]
+}

Solutions/Censys Supplier Risk/Analytic Rules/CensysRiskStateChange.yaml

@@ -0,0 +1,102 @@
+id: 8a208922-5ef7-4721-ad8b-7f69e2163d79
+name: Censys SR - Risk State Change
+version: 1.0.0
+kind: Scheduled
+status: Available
+requiredDataConnectors:
+  - connectorId: CensysSRRiskEventsFunction
+    dataTypes:
+      - CensysRiskEvents_CL
+queryFrequency: 1h
+queryPeriod: 1d
+suppressionDuration: PT1H
+suppressionEnabled: false
+triggerOperator: GreaterThan
+triggerThreshold: 0
+tactics:
+  - InitialAccess
+relevantTechniques:
+  - T1195
+severity: Medium
+description: |
+  Creates one incident per supplier risk instance that is opened, or reopened, with CRITICAL or HIGH severity. Alerts are grouped by RiskInstanceId, so further events of the same risk instance within 24 hours are added to the same incident while it is open.
+  Alert severity follows the Censys severity: CRITICAL and HIGH map to High, MEDIUM to Medium and LOW to Low. Resolved events follow the same mapping, and any other severity is Informational.
+  By default the rule alerts only on new CRITICAL and HIGH risks. The query holds commented lines to follow the whole life of a risk instead (reopened events of any severity and resolved events), see the comments in the query.
+query: |
+  // New events are selected by the time they reached Sentinel (IngestedAt), not by the Censys insert time,
+  // so a delay between Censys and the connector can never make an event fall between two runs.
+  // Each run covers 60 minutes that end 5 minutes ago, and consecutive runs are contiguous, so no event is alerted twice.
+  //
+  // OPTIONAL - follow the whole life of a risk instance (reopened events of any severity, and resolved events of risks that
+  // were alerted before). To use it, uncomment the two "let" lines below, comment out the line "Default" in the query, uncomment
+  // the line "Full lifecycle", and in the rule settings set the query period to 7 days and the incident grouping look-back to 7 days.
+  // let Alertable = materialize(CensysRiskEvents | where ChangeType == "reopened" or (ChangeType == "opened" and Severity in ("CRITICAL", "HIGH")));
+  // let TrackedRisks = Alertable | distinct RiskInstanceId;
+  CensysRiskEvents
+  | where IngestedAt between (ago(65m) .. ago(5m))
+  // Default: opened or reopened risks with CRITICAL or HIGH severity
+  | where ChangeType in ("opened", "reopened") and Severity in ("CRITICAL", "HIGH")
+  // Full lifecycle (needs the optional "let" lines above):
+  // | where (ChangeType == "reopened" or (ChangeType == "opened" and Severity in ("CRITICAL", "HIGH"))) or (ChangeType == "resolved" and RiskInstanceId in (TrackedRisks))
+  //
+  // Optional filters: remove the leading // of a line to use it. Each line narrows the alerts further, and several lines can be combined.
+  // Do not remove or change the IngestedAt window above, it keeps every event alerted exactly once.
+  // | where Confidence >= 0.8                           // only findings with a confidence of 80% or more
+  // | where Severity == "CRITICAL"                      // only CRITICAL risks (Censys severities: CRITICAL, HIGH, MEDIUM, LOW)
+  // | where ChangeType == "opened"                      // only newly opened risks, skip reopened risks
+  // | where RapidResponse == true                       // only actively exploited (rapid response) risks
+  // | where RiskCategory == "VULNERABILITY"             // VULNERABILITY, MISCONFIGURATION, EXPOSURE or THREAT
+  // | where SupplierDisplayName in ("OpenAI", "Harvey") // only these suppliers
+  // | where CveId != ""                                 // only risks that have a CVE id
+  // | where RiskName has "Log4Shell"                    // only risks whose name contains this text
+  // | where AssetType == "host"                         // asset type: host or web
+  | extend SentinelSeverity = case(
+      Severity in ("CRITICAL", "HIGH"), "High",
+      Severity == "MEDIUM", "Medium",
+      Severity == "LOW", "Low",
+      "Informational")
+  | project TimeGenerated, IngestedAt, EventId, InsertedAt, OccurredAt, ChangeType, RiskInstanceId, Severity, SentinelSeverity,
+      RiskCategory, RiskName, Description, CveId, RapidResponse, SupplierDisplayName,
+      AssetName, AssetHostname, AssetIp, ServicePort, ServiceProtocol, Evidence, ResolveTime, ResolveReason, PlatformUrl
+entityMappings:
+  - entityType: Host
+    fieldMappings:
+      - identifier: HostName
+        columnName: AssetHostname
+  - entityType: IP
+    fieldMappings:
+      - identifier: Address
+        columnName: AssetIp
+  - entityType: URL
+    fieldMappings:
+      - identifier: Url
+        columnName: PlatformUrl
+eventGroupingSettings:
+  aggregationKind: AlertPerResult
+incidentConfiguration:
+  createIncident: true
+  groupingConfiguration:
+    enabled: true
+    reopenClosedIncident: false
+    lookbackDuration: P1D
+    matchingMethod: Selected
+    groupByCustomDetails:
+      - RiskInstanceId
+customDetails:
+  RiskInstanceId: RiskInstanceId
+  ChangeType: ChangeType
+  CensysSeverity: Severity
+  RiskCategory: RiskCategory
+  Supplier: SupplierDisplayName
+  CveId: CveId
+  RapidResponse: RapidResponse
+  Evidence: Evidence
+  ResolveReason: ResolveReason
+  ResolveTime: ResolveTime
+alertDetailsOverride:
+  alertDisplayNameFormat: "Censys: {{ChangeType}} risk - {{RiskName}} ({{SupplierDisplayName}})"
+  alertDescriptionFormat: "{{Description}} Asset: {{AssetName}}. Open in Censys: {{PlatformUrl}}"
+  alertSeverityColumnName: SentinelSeverity
+  alertDynamicProperties:
+    - alertProperty: AlertLink
+      value: PlatformUrl

Solutions/Censys Supplier Risk/Data Connectors/CensysRiskEventsFunctionConnector/CensysSRRiskEvents/__init__.py

@@ -0,0 +1,129 @@
+"""Timer triggered function: ingest Censys risk events into Microsoft Sentinel."""
+import datetime
+import time
+from typing import List, Optional
+
+import azure.functions as func
+
+from ..SharedCode import consts
+from ..SharedCode.censys_client import CensysClient
+from ..SharedCode.checkpoint import CheckpointManager, checkpoint_key
+from ..SharedCode.exceptions import CensysCursorExpired, CensysException, CensysTimeBudgetReached
+from ..SharedCode.logger import applogger
+from ..SharedCode.sentinel import SentinelIngestor
+from ..SharedCode.suppliers import RiskStore, SupplierStore, resolve
+from ..SharedCode.validation import Settings, load_settings
+
+_RFC3339 = "%Y-%m-%dT%H:%M:%SZ"
+
+
+def _supplier_ids(settings: Settings) -> Optional[List[str]]:
+    """Look up the ids of the supplier names, once per run. Returns None when no supplier names are set.
+
+    A supplier added to Censys later is used from the next run on, without reading its earlier events. A name that
+    matches no supplier is a warning, not an error. An empty list means that no id was found at all.
+    """
+    if not settings.supplier_names:
+        return None
+    store = SupplierStore(settings.storage_connection, consts.SUPPLIERS_TABLE_NAME, settings.organization_id)
+    ids, missing = resolve(settings.supplier_names, store.load())
+    for name in missing:
+        applogger.warning(f"{consts.LOG_PREFIX} ingest: supplier name '{name}' was not found "
+                          "in the suppliers table, it is skipped")
+    if len(ids) > consts.MAX_SUPPLIER_IDS:
+        raise CensysException(f"The supplier names in SupplierNames match {len(ids)} suppliers, at most "
+                              f"{consts.MAX_SUPPLIER_IDS} are allowed in one request. Enter fewer names.")
+    return ids
+
+
+def _risk_type_ids(settings: Settings) -> Optional[List[str]]:
+    """Look up the ids of the risk names, once per run. Returns None when no risk names are set.
+
+    Works like _supplier_ids: an unknown name is a warning, and an empty list means that no id was found at all.
+    """
+    if not settings.risk_names:
+        return None
+    store = RiskStore(settings.storage_connection, consts.RISKS_TABLE_NAME, settings.organization_id)
+    ids, missing = resolve(settings.risk_names, store.load())
+    for name in missing:
+        applogger.warning(f"{consts.LOG_PREFIX} ingest: risk name '{name}' was not found "
+                          "in the risks table, it is skipped")
+    if len(ids) > consts.MAX_RISK_TYPE_IDS:
+        raise CensysException(f"The risk names in RiskNames match {len(ids)} risks, at most "
+                              f"{consts.MAX_RISK_TYPE_IDS} are allowed in one request. Enter fewer names.")
+    return ids
+
+
+def ingest(start: float) -> int:
+    """Read every new event, upload it page by page and move the checkpoint after each page."""
+    settings = load_settings()
+    supplier_ids = _supplier_ids(settings)
+    risk_type_ids = _risk_type_ids(settings)
+    for setting, ids in (("SupplierNames", supplier_ids), ("RiskNames", risk_type_ids)):
+        if ids is not None and not ids:
+            applogger.warning(f"{consts.LOG_PREFIX} ingest: no id found for the names set in {setting}, "
+                              "nothing is collected in this run. The next run tries again.")
+            return 0
+    filters = settings.filters(supplier_ids, risk_type_ids)
+    # The suppliers and risks are not part of the checkpoint key. When their names are changed, collection continues
+    # from the current checkpoint with the new ids, and the earlier events of a newly selected supplier or risk
+    # are not collected.
+    key_parts = [f"{k}={v}" for k, v in sorted(filters.items()) if k not in ("supplier_ids", "risk_type_ids")]
+    key = checkpoint_key(settings.organization_id, *key_parts)
+    manager = CheckpointManager(settings.storage_connection, consts.CHECKPOINT_TABLE_NAME)
+    client = CensysClient(settings.api_token, settings.organization_id, deadline=start + consts.MAX_RUN_SECONDS,
+                          hard_deadline=start + consts.HARD_LIMIT_SECONDS)
+    ingestor = SentinelIngestor(settings.data_collection_endpoint, settings.data_collection_rule_id,
+                                consts.STREAM_NAME, consts.SCOPE)
+
+    checkpoint = manager.get(key)
+    if checkpoint and checkpoint[0]:
+        after_time, after_id = checkpoint
+        applogger.info(f"{consts.LOG_PREFIX} ingest: continuing from checkpoint "
+                       f"inserted_at={after_time} id={after_id}")
+    else:
+        after_time, after_id = settings.start_time, None
+        applogger.info(f"{consts.LOG_PREFIX} ingest: no checkpoint, starting from {after_time}")
+
+    total, expired = 0, False
+    while True:
+        try:
+            for events in client.iter_pages(after_time, after_id, filters):
+                ingestor.upload(events)
+                total += len(events)
+                last = events[-1]
+                if last.get("inserted_at") and last.get("id"):
+                    manager.set(key, last["inserted_at"], last["id"])
+                else:
+                    applogger.warning(f"{consts.LOG_PREFIX} ingest: last event of the page has no inserted_at "
+                                      "or id, checkpoint not moved")
+                applogger.info(f"{consts.LOG_PREFIX} ingest: ingested {len(events)} events, "
+                               f"checkpoint inserted_at={last.get('inserted_at')}")
+            return total
+        except CensysTimeBudgetReached:
+            applogger.info(f"{consts.LOG_PREFIX} ingest: time budget of {consts.MAX_RUN_SECONDS} seconds reached "
+                           f"after {total} events. The checkpoint is saved and the next run continues from it.")
+            return total
+        except CensysCursorExpired:
+            if expired:
+                raise
+            expired = True
+            window = datetime.timedelta(days=consts.RETENTION_DAYS)
+            after_time = (datetime.datetime.now(datetime.timezone.utc) - window).strftime(_RFC3339)
+            after_id = None
+            applogger.warning(f"{consts.LOG_PREFIX} ingest: the checkpoint is older than the retention window, "
+                              f"restarting from {after_time}")
+
+
+def main(mytimer: func.TimerRequest) -> None:
+    """Entry point called by the timer trigger."""
+    start = time.time()
+    applogger.info(f"{consts.LOG_PREFIX} main: start processing")
+    if mytimer.past_due:
+        applogger.info(f"{consts.LOG_PREFIX} main: the timer is past due")
+    try:
+        total = ingest(start)
+    except CensysException as err:
+        applogger.error(f"{consts.LOG_PREFIX} main: run failed: {err}")
+        raise
+    applogger.info(f"{consts.LOG_PREFIX} main: finished, {total} events ingested in {int(time.time() - start)} sec")

Solutions/Censys Supplier Risk/Data Connectors/CensysRiskEventsFunctionConnector/CensysSRRiskEvents/function.json

@@ -0,0 +1,11 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "name": "mytimer",
+      "type": "timerTrigger",
+      "direction": "in",
+      "schedule": "%Schedule%"
+    }
+  ]
+}