I analyzed VPC flow logs with DuckDB and narrowed down security groups with 0.0.0.0/0

Hello, I'm Hayashi.

For a certain project, I worked on restricting security groups that were open with 0.0.0.0/0.

Since we can't just restrict them immediately and cut off communication, we first confirm the "ranges actually in use" from the real traffic recorded in VPC flow logs before tightening the rules.

Athena is the standard tool for flow log analysis.
However, since this environment is a customer's AWS account, we wanted to avoid creating resources like tables just for analysis, or incurring charges that vary depending on the scan volume.
So, we retrieved the logs locally and analyzed them with DuckDB.

This article introduces the steps from identifying the permitted ranges from traffic recorded in flow logs to remediation.

For the basics of flow log analysis with DuckDB, I referenced this article:
https://dev.classmethod.jp/articles/query-vpc-flow-logs-by-duckdb/

Environment Overview

An EC2 instance running an internal BI tool had the following security group attached.

This article proceeds on the assumption that the following two things are ready.

• VPC flow logs are being output to S3
• An environment that can access S3 (AWS CloudShell in this case) and a local environment to run DuckDB

We'll proceed in the following steps.

1. Identify the ENI that has the target security group attached
2. Retrieve flow logs from S3 to local
3. Load them into DuckDB
4. Narrow down the target, convert to Parquet, and aggregate ACCEPTED traffic
5. Create a rule proposal from the aggregation results and restrict the security group
6. Check whether any REJECTs appear after restricting

Step 1: Identify the ENI with the Security Group Attached

Security group IDs are not recorded in flow logs. What is recorded is at the ENI level.

First, find out which ENIs the target security group is attached to.
At the same time, also check what security groups are attached to each ENI.

aws ec2 describe-network-interfaces \
  --filters Name=group-id,Values=<security group ID> \
  --query "NetworkInterfaces[].{eni:NetworkInterfaceId,ip:PrivateIpAddress,sgs:Groups[].GroupId}"

[
    {
        "eni": "eni-0aaaa1111bbbb2222c",
        "ip": "10.0.123.13",
        "sgs": ["<security group ID>", "sg-0111111111aaaaaaa", "sg-0222222222bbbbbbb"]
    },
    {
        "eni": "eni-0dddd3333eeee4444f",
        "ip": "10.0.223.13",
        "sgs": ["<security group ID>", "sg-0333333333ccccccc"]
    }
]

As shown in the output, a single ENI can have multiple security groups attached. Since flow logs are recorded at the ENI level, it's not possible to determine which security group permitted a specific connection.

The clue we use here is the destination port (dstport). In this case, since the target security group was the one opening ports 8080/9001, we can extract traffic destined for these ports as the relevant communications. The subsequent steps filter by these ports.

Step 2: Retrieve Flow Logs Locally

Log retrieval from S3 was done in AWS CloudShell, and analysis was done in the local environment. CloudShell is a browser-accessible terminal with AWS CLI pre-authenticated, so you can access S3 directly with the aws command without preparing credentials locally.

First, retrieve the logs for the period you want to analyze into CloudShell.

aws s3 sync "s3://<flow log bucket>/AWSLogs/123456789012/vpcflowlogs/ap-northeast-1/" ./flowlogs/ \
  --exclude "*" --include "*2026-06*" --include "*2026-07*"

Flow logs are output in directories separated by date, so you can narrow down to the target period using --include.

s3://<bucket>/AWSLogs/123456789012/vpcflowlogs/ap-northeast-1/
├── 2026/06/10/
│   └── 123456789012_vpcflowlogs_ap-northeast-1_fl-0abc..._20260610T0000Z_xxxx.log.gz
├── 2026/06/11/
│   └── ...
└── 2026/07/10/
    └── ...

The approximately one month of data we targeted this time was about 100MB total in gzip-compressed form.

The retrieved logs can be downloaded locally from CloudShell's "Actions" → "Download file". Since there are many files, bundling them with tar into a single file before downloading makes it easier.

tar czf flowlogs.tar.gz flowlogs/

Step 3: Load into DuckDB

From here, we work locally. Extract the downloaded logs.

tar xzf flowlogs.tar.gz

If DuckDB is not installed, install it with the official installation script.
(DuckDB Installation)

curl https://install.duckdb.org | sh

After installation, typing duckdb launches an interactive prompt where you can enter and execute SQL directly. If you collect SQL into a file, you can also run it all at once like duckdb -c ".read analysis.sql".

From here, we'll set up a state where we can query the log files with SQL.
There are two things to do.

1. Assign names to each column of the log (make them SQL-friendly, like action or srcaddr)
2. Extract only the data needed for analysis and save it in a convenient format

Assigning Names to Each Column of the Log

Each line of the flow log is space-delimited text with no column names.
First, confirm what field is at which position.

For the default format, the field order is fixed, but for custom formats, it varies by environment. Check with describe-flow-logs.

aws ec2 describe-flow-logs --query "FlowLogs[].LogFormat"

In this environment, it was a custom format with 29 fields like this.
(This is the format you get when selecting "All attributes" when creating flow logs.)

[
  "${account-id} ${action} ${az-id} ${bytes} ${dstaddr} ${dstport} ${end} ${flow-direction} ${instance-id} ${interface-id} ${log-status} ${packets} ${pkt-dst-aws-service} ${pkt-dstaddr} ${pkt-src-aws-service} ${pkt-srcaddr} ${protocol} ${region} ${srcaddr} ${srcport} ${start} ${sublocation-id} ${sublocation-type} ${subnet-id} ${tcp-flags} ${traffic-path} ${type} ${version} ${vpc-id}"
]

When DuckDB reads a file without column names, it automatically names each column from the beginning as column00, column01, ... The order of the fields above corresponds to these numbers, so we create a "view" with renamed columns for only the necessary ones.
By creating a view, subsequent queries can reference it by the name flow_logs without writing out the long read specification.

CREATE OR REPLACE VIEW flow_logs AS
SELECT
  column01 AS action,        -- ACCEPT / REJECT
  CAST(column03 AS BIGINT) AS bytes,
  column04 AS dstaddr,       -- Destination IP
  TRY_CAST(column05 AS INT) AS dstport,   -- Destination port
  column07 AS flow_direction,
  column09 AS interface_id,  -- ENI
  column10 AS log_status,
  TRY_CAST(column16 AS INT) AS protocol,
  column18 AS srcaddr,       -- Source IP
  to_timestamp(CAST(column20 AS BIGINT)) AS start_time
  -- Please adjust to match the actual LogFormat
FROM read_csv('flowlogs/**/*.log.gz', delim=' ', header=false, all_varchar=true);

From top to bottom in the SQL, here are the meanings of the conversion operations and read options used.

• CAST(...): A SQL operation that converts strings to numeric types, used for columns used in aggregation or comparison
• TRY_CAST(...): A safe version of CAST that treats unconvertible values like - (no record) as empty (NULL) instead of erroring (necessary for columns like ports)
• to_timestamp(...): Converts timestamps recorded as UNIX seconds into readable date/time
• FROM read_csv(...): Reads all .log.gz files under flowlogs/ at once (header=false means no header row, all_varchar=true means read all columns as strings)

<version> <account-id> <interface-id> <srcaddr> <dstaddr> <srcport> <dstport> <protocol> <packets> <bytes> <start> <end> <action> <log-status>

CREATE OR REPLACE VIEW flow_logs AS
SELECT
  column02 AS interface_id,  -- ENI
  column03 AS srcaddr,       -- Source IP
  column04 AS dstaddr,       -- Destination IP
  TRY_CAST(column06 AS INT) AS dstport,   -- Destination port
  TRY_CAST(column07 AS INT) AS protocol,
  CAST(column09 AS BIGINT) AS bytes,
  to_timestamp(CAST(column10 AS BIGINT)) AS start_time,
  column12 AS action,        -- ACCEPT / REJECT
  column13 AS log_status
FROM read_csv('flowlogs/**/*.log.gz', delim=' ', header=false, all_varchar=true);

Extract Only the Analysis Target and Save as Parquet

Reading the entire log every time takes time.
So, we extract only the range we want to analyze this time (inbound traffic to ports 8080/9001 of the target ENIs) and save it as a file called target_flows.parquet.
Parquet is a format that can be read quickly by column, and subsequent queries are executed against this file.

COPY (
  SELECT * FROM flow_logs
  WHERE flow_direction = 'ingress'   -- Inbound direction
    AND log_status = 'OK'            -- Has records
    AND interface_id IN ('eni-0aaaa1111bbbb2222c', 'eni-0dddd3333eeee4444f')
    AND protocol = 6                 -- TCP
    AND dstport IN (8080, 9001)
) TO 'target_flows.parquet' (FORMAT PARQUET);

The conditions specified in WHERE have the following meanings.

• flow_direction = 'ingress': Narrow down to inbound traffic coming from outside
• log_status = 'OK': Exclude NODATA / SKIPDATA (rows that are not records of actual traffic)
• interface_id IN (...): Limit to the target ENIs identified in Step 1
• protocol = 6: Limit to TCP (6 is the protocol number for TCP)
• dstport IN (8080, 9001): Narrow down to traffic destined for the ports opened by the target security group

Note that each line in the flow log is not an individual packet, but a "flow" unit that aggregates similar traffic within a certain period of time. The count referred to here is the number of these flows.

As a result of this filtering, 212,158 flows were targeted this time.
From reading approximately 100MB of logs to outputting target_flows.parquet took about 1.5 seconds.
Subsequent aggregations run against this pre-filtered file, so they'll be even faster.

Step 4: Aggregate ACCEPTED Traffic

First, aggregate permitted traffic (action = 'ACCEPT') by combination of source IP and destination port.
Let's look at the top 10 results, ordered by number of flows.

SELECT srcaddr, dstport, count(*) AS flows, sum(bytes) AS total_bytes
FROM 'target_flows.parquet'
WHERE action = 'ACCEPT'
GROUP BY srcaddr, dstport
ORDER BY flows DESC
LIMIT 10;

We filter to only permitted traffic with action = 'ACCEPT' and aggregate the number of flows (count(*)) and traffic volume (sum(bytes)) for each combination of source IP and destination port.
Display the top 10 in descending order by count.

The top results were all internal network IPs starting with 10.100.. However, there are many varieties of source IPs, and looking at just the top 10 doesn't give us the full picture.
So, we count "how many unique source IPs there are" per ENI and port. The deduplicated count can be obtained with count(DISTINCT srcaddr).

SELECT f.interface_id, f.dstport, count(*) AS flows,
       count(DISTINCT f.srcaddr) AS unique_sources
FROM 'target_flows.parquet' f
WHERE f.action = 'ACCEPT'
GROUP BY ALL
ORDER BY flows DESC;

Port 8080 on the production side (eni-0aaaa1111bbbb2222c) recorded over 200,000 flows from approximately 2,500 unique sources over one month. Checking each source IP one by one is not realistic, so in the next step we'll organize them by grouping into network ranges.

Step 5: Create a Rule Proposal from the Aggregation Results

Rather than looking at each source IP individually, we aggregate by grouping them by internal network classification (internal NW, VPN, AWS environment).

SELECT dstport,
       CASE
         WHEN srcaddr LIKE '10.100.%' THEN '10.100.0.0/16 (Internal NW)'
         WHEN srcaddr LIKE '10.200.%' THEN '10.200.0.0/16 (VPN A)'
         WHEN srcaddr LIKE '10.210.%' THEN '10.210.0.0/16 (VPN B)'
         WHEN srcaddr LIKE '10.0.%'   THEN '10.0.0.0/16 (AWS environment)'
         ELSE 'Other'
       END AS src_range,
       count(*) AS flows, count(DISTINCT srcaddr) AS unique_ips
FROM 'target_flows.parquet'
WHERE action = 'ACCEPT'
GROUP BY ALL ORDER BY dstport, flows DESC;

The CASE expression classifies source IPs by network.
(LIKE '10.100.%' means "IPs starting with 10.100.", and anything that doesn't match any range is grouped into "Other" with ELSE.)

The notable finding here is that "Other" had 0 entries.
Although it was open with 0.0.0.0/0, all traffic ACCEPTED during the one-month period came from internal, VPN, and AWS IP ranges — there was no access from the internet.

This means that by allowing only these 4 ranges, we can eliminate 0.0.0.0/0 without interrupting actual traffic.
The proposed rules are as follows.

Checkpoints before restricting:

• Does the log observation period cover at least one full business cycle?
  (Low-frequency traffic like monthly batch jobs won't appear in short periods, so we checked one month's worth this time)
• If source addresses are limited to very few IPs, there's also the option of restricting to /32 (that IP only) rather than a range like /16
  (The AWS environment for port 9001 this time had only 1 IP)
• Note that traffic through ELB or NAT may appear with different source IPs

After that, delete the 0.0.0.0/0 rule and add rules for the necessary ranges according to this proposal, and remediation is complete. After remediation, verify the results with describe-security-group-rules.

aws ec2 describe-security-group-rules \
  --filters Name=group-id,Values=<security group ID> \
  --query "sort_by(SecurityGroupRules[?IsEgress==\`false\`],&FromPort)[].{proto:IpProtocol,port:FromPort,cidr:CidrIpv4,desc:Description}"

[
    {"proto": "tcp", "port": 8080, "cidr": "10.100.0.0/16", "desc": "Internal NW"},
    {"proto": "tcp", "port": 8080, "cidr": "10.200.0.0/16", "desc": "VPN A"},
    {"proto": "tcp", "port": 8080, "cidr": "10.210.0.0/16", "desc": "VPN B"},
    {"proto": "tcp", "port": 8080, "cidr": "10.0.0.0/16", "desc": "AWS"},
    {"proto": "tcp", "port": 9001, "cidr": "10.100.0.0/16", "desc": "Internal NW"},
    {"proto": "tcp", "port": 9001, "cidr": "10.0.0.0/16", "desc": "AWS"}
]

0.0.0.0/0 is gone, and only the ranges with actual traffic remain.

Verification After Remediation

After restricting the security group, it's reassuring to check for rejected traffic (REJECT) after operating for a while. If legitimate traffic was accidentally blocked, it will appear here as REJECT.
Retrieve the logs for the post-remediation period using the same approach as Steps 2-3, recreate the view (flow_logs), and then run the following query.

SELECT srcaddr, dstport, count(*) AS rejects
FROM flow_logs
WHERE action = 'REJECT' AND flow_direction = 'ingress'
GROUP BY srcaddr, dstport
ORDER BY rejects DESC;

We filter to only rejected traffic with action = 'REJECT' and count by source IP and destination port. If unexpected REJECTs appear, check whether the source is legitimate and add it to the rules if necessary. The post-remediation verification can also be done with just the same DuckDB aggregation as before.

Comparison with Athena

With about 100MB this time, even Athena would cost less than 0.1 yen per query, so the cost difference was minimal. Still, by using DuckDB, we avoided creating analysis resources in the customer's account and didn't have to worry about scan charges. After retrieving the logs, no matter how many times we rerun queries, there's no impact on the customer's environment.

Summary

• When restricting 0.0.0.0/0, looking at ACCEPTED traffic in flow logs lets you determine the necessary permitted ranges from data
• For one-time analysis like audits, DuckDB is sufficient (about 100MB in about 1.5 seconds, negligible cost)
• You can analyze without creating resources in the customer's account and without worrying about charges

Here are some pitfalls to watch out for during analysis.

• Rows with dstport = 0 are ICMP (becomes 0 because there's no concept of ports), so check them together with the protocol column
• ENIs and security groups are not 1-to-1, so matching which traffic was permitted by which rule requires cross-referencing ports and rules
• Custom format logs require column mapping (see Step 3)
• Flow log REJECTs cannot distinguish whether traffic was denied by a security group or a NACL

Closing Thoughts

Recently, it's become possible to delegate everything from this kind of analysis to creating remediation proposals to AI.
This time, since I wanted to understand the mechanism myself, I deliberately went through it hands-on.

I hope this article is helpful to someone. Thank you for reading to the end!

References

https://dev.classmethod.jp/articles/query-vpc-flow-logs-by-duckdb/

https://dev.classmethod.jp/articles/vpc-flow-log-default-field/

https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html