Security groups and NSG rules define what should be allowed and blocked. Without flow logs, you have no evidence of what traffic actually reached your instances, which blocked connections were attempted, and whether east-west traffic inside your VCN is behaving unexpectedly. OCI VCN Flow Logs capture metadata for every IP flow: source and destination IP, port, protocol, bytes, packets, and ACCEPT or REJECT. This post covers enabling them with Terraform, writing security detection queries in Logging Analytics LQL, and wiring automated alarms.

Step 1: Enable Flow Logs on Production Subnets

resource "oci_logging_log_group" "vcn_flow_logs" {
  compartment_id = var.compartment_id
  display_name   = "vcn-flow-log-group"
  description    = "VCN flow logs for all production subnets"
}

resource "oci_logging_log" "app_subnet_flow" {
  display_name = "app-subnet-flow-log"
  log_group_id = oci_logging_log_group.vcn_flow_logs.id
  log_type     = "SERVICE"

  configuration {
    source {
      category    = "all"
      resource    = var.app_subnet_id
      service     = "flowlogs"
      source_type = "OCISERVICE"
    }
    compartment_id = var.compartment_id
  }

  retention_duration = 90
  is_enabled         = true
}

resource "oci_logging_log" "db_subnet_flow" {
  display_name = "db-subnet-flow-log"
  log_group_id = oci_logging_log_group.vcn_flow_logs.id
  log_type     = "SERVICE"

  configuration {
    source {
      category    = "all"
      resource    = var.db_subnet_id
      service     = "flowlogs"
      source_type = "OCISERVICE"
    }
    compartment_id = var.compartment_id
  }

  retention_duration = 90
  is_enabled         = true
}

resource "oci_sch_service_connector" "flow_to_la" {
  compartment_id = var.compartment_id
  display_name   = "vcn-flow-to-logging-analytics"

  source {
    kind = "logging"
    log_sources {
      compartment_id = var.compartment_id
      log_group_id   = oci_logging_log_group.vcn_flow_logs.id
    }
  }

  target {
    kind                  = "loggingAnalytics"
    log_group_id          = var.la_infrastructure_log_group_id
    log_source_identifier = "OCI VCN Flow Unified Schema Logs"
  }
}

Step 2: Security Detection Queries in LQL

-- Port scan detection: source IP hitting many ports in 5 minutes
'Log Source' = 'OCI VCN Flow Unified Schema Logs'
and Action = 'REJECT'
| eval time_bucket = bin('Event Time', 5minutes)
| stats
    count()                            as total_attempts,
    count_distinct('Destination Port') as unique_ports
    by 'Source IP', time_bucket
| where total_attempts > 20 and unique_ports > 10
| sort -total_attempts

-- Lateral movement: instance connecting to many private IPs
'Log Source' = 'OCI VCN Flow Unified Schema Logs'
and Action = 'ACCEPT'
and 'Destination IP' like '10.%'
| eval time_bucket = bin('Event Time', 10minutes)
| stats count_distinct('Destination IP') as unique_destinations
  by 'Source IP', time_bucket
| where unique_destinations > 15
| sort -unique_destinations

-- Data exfiltration: large outbound bytes to external IPs
'Log Source' = 'OCI VCN Flow Unified Schema Logs'
and Action = 'ACCEPT'
and 'Destination IP' not like '10.%'
and 'Destination IP' not like '192.168.%'
| timestats sum('Bytes Sent') as total_bytes span = 5minutes by 'Source IP'
| where total_bytes > 100000000
| sort -total_bytes

-- Unexpected database access: non-app-subnet connections to port 1521
'Log Source' = 'OCI VCN Flow Unified Schema Logs'
and 'Destination Port' = '1521'
and 'Source IP' not like '10.0.1.%'
| stats count() as attempts by 'Source IP', Action
| sort -attempts

Step 3: Automated Security Alert

resource "oci_log_analytics_scheduled_task" "port_scan_detection" {
  compartment_id = var.compartment_id
  namespace      = var.tenancy_namespace
  display_name   = "Port Scan Detection"
  kind           = "STANDARD"
  task_type      = "SAVED_SEARCH"

  action {
    type            = "STREAM"
    saved_search_id = oci_log_analytics_log_analytics_saved_search.port_scan.id

    metric_extraction {
      compartment_id = var.compartment_id
      namespace      = "custom_security_metrics"
      metric_name    = "PortScanAttempts"
      resource_group = "network_security"
    }
  }

  schedules {
    schedule {
      type           = "CRON"
      expression     = "0 */5 * * * ?"
      time_zone      = "UTC"
      misfire_policy = "RETRY_ONCE"
    }
  }
}

resource "oci_monitoring_alarm" "port_scan_alarm" {
  compartment_id        = var.compartment_id
  display_name          = "port-scan-detected"
  is_enabled            = true
  metric_compartment_id = var.compartment_id
  namespace             = "custom_security_metrics"
  query                 = "PortScanAttempts[5m]{resourceGroup = 'network_security'}.max() > 0"
  severity              = "CRITICAL"
  pending_duration      = "PT5M"
  destinations          = [var.security_notification_topic_id]
  body                  = "Port scan pattern detected in VCN flow logs. Review the source IP in Logging Analytics immediately."
}

Step 4: Python Flow Log Analyzer

import oci, json
from datetime import datetime, timezone, timedelta
from collections import defaultdict

def analyze_rejected_connections(log_group_id: str, hours: int = 1):
    config = oci.config.from_file()
    search = oci.loggingsearch.LogSearchClient(config)
    now    = datetime.now(timezone.utc)

    resp = search.search_logs(
        search_logs_details=oci.loggingsearch.models.SearchLogsDetails(
            time_start=now - timedelta(hours=hours),
            time_end=now,
            search_query='search "' + log_group_id + '" | where data.action = REJECT',
            is_return_field_info=False
        ),
        limit=1000
    )

    by_src = defaultdict(lambda: {"count": 0, "ports": set()})
    for result in resp.data.results:
        d = json.loads(result.data) if isinstance(result.data, str) else result.data
        src  = d.get("srcaddr", "unknown")
        port = d.get("dstport", 0)
        by_src[src]["count"] += 1
        by_src[src]["ports"].add(port)

    scanners = {ip: info for ip, info in by_src.items()
                if info["count"] > 20 and len(info["ports"]) > 10}

    print(f"Total rejected flows: {sum(i['count'] for i in by_src.values())}")
    print(f"Potential scanners:   {len(scanners)}")
    for ip, info in scanners.items():
        print(f"  {ip}: {info['count']} attempts across {len(info['ports'])} ports")

Operational Notes

Flow logs capture at the subnet boundary. Traffic between two instances in the same subnet may not appear in flow logs if it does not cross a subnet boundary. For intra-subnet monitoring, use application-level access logs or OS-level auditing on the instances themselves.

Set retention to at least 90 days. Security investigations frequently look back weeks or months to trace the origin of an incident. Route flow logs to Object Storage via Service Connector Hub for archival beyond the Logging retention limit, and apply a lifecycle policy to transition older logs to the Archive storage tier.

Regards,
Osama

#OCI #OracleCloud #VCN #FlowLogs #CloudSecurity #Terraform #TechBlog #Oracle #NetworkSecurity #ThreatDetection #LoggingAnalytics #NetworkMonitoring #OracleCloudInfrastructure #IaC #DevSecOps #SIEM #SecurityAnalytics #PlatformEngineering #CloudNetworking #ZeroTrust