The hardware and bandwidth for this mirror is donated by dogado GmbH, the Webhosting and Full Service-Cloud Provider. Check out our Wordpress Tutorial.
If you wish to report a bug, or if you are interested in having us mirror your free-software or open-source project, please feel free to contact us at mirror[@]dogado.de.

Network Traffic Analysis from AstraeaDB

Generated with R-AstraeaDB

2026-08-19

Overview

This document analyses IP traffic that was captured by pcap_to_graph.R and stored in AstraeaDB. The graph model is:

library(AstraeaDB)
library(igraph)
library(ggplot2)
library(data.table)
library(scales)
library(visNetwork)

# Helper: flatten a list column to an atomic vector (NULL -> NA)
unbox_col <- function(x) {
  sapply(x, function(el) if (is.null(el) || length(el) == 0) NA else el[[1]])
}

1 – Data Extraction

Connect to AstraeaDB, discover every IP node, and collect all flow edges.

client <- astraea_connect()
client$ping()
# Use GQL to find all IPAddress nodes regardless of their internal IDs.
node_result <- client$query("MATCH (n:IPAddress) RETURN n")
nodes_list <- lapply(node_result$rows, function(row) {
  n <- row[[1]]
  list(node_id = n$id, ip = n$properties$ip %||% NA_character_)
})

nodes_dt <- rbindlist(nodes_list, fill = TRUE)

# Flatten list columns to atomic types
for (col in names(nodes_dt)) {
  if (is.list(nodes_dt[[col]])) {
    set(nodes_dt, j = col, value = unbox_col(nodes_dt[[col]]))
  }
}
nodes_dt[, node_id := as.character(node_id)]
nodes_dt[, ip := as.character(ip)]

cat(sprintf("Discovered %d IP address nodes.\n", nrow(nodes_dt)))
# Use GQL to pull all flow edges between IPAddress nodes.
edge_result <- client$query(
  "MATCH (s:IPAddress)-[e]->(d:IPAddress)
   RETURN id(e), s.ip, d.ip, type(e), e.service_port, e.service_name,
          e.flow_count, e.total_bytes"
)

edges_list <- lapply(edge_result$rows, function(row) {
  list(
    edge_id      = row[[1]],
    src_ip       = row[[2]],
    dst_ip       = row[[3]],
    protocol     = row[[4]] %||% "IP",
    service_port = row[[5]] %||% NA_integer_,
    service_name = row[[6]] %||% NA_character_,
    flow_count   = row[[7]] %||% 1L,
    total_bytes  = row[[8]] %||% NA_integer_
  )
})

edges_dt <- rbindlist(edges_list, fill = TRUE)

# Flatten list columns to atomic types (query results may wrap values in lists)
for (col in names(edges_dt)) {
  if (is.list(edges_dt[[col]])) {
    set(edges_dt, j = col, value = unbox_col(edges_dt[[col]]))
  }
}
# Coerce to expected types
edges_dt[, edge_id      := as.character(edge_id)]
edges_dt[, src_ip       := as.character(src_ip)]
edges_dt[, dst_ip       := as.character(dst_ip)]
edges_dt[, protocol     := as.character(protocol)]
edges_dt[, service_port := as.integer(service_port)]
edges_dt[, service_name := as.character(service_name)]
edges_dt[, flow_count   := as.integer(flow_count)]
edges_dt[, total_bytes  := as.numeric(total_bytes)]

cat(sprintf("Collected %d flow edges (representing %s packets).\n",
            nrow(edges_dt),
            format(sum(edges_dt$flow_count, na.rm = TRUE), big.mark = ",")))
# Timestamps live in the edge's valid_from / valid_to fields.
edges_dt[, c("valid_from_ms", "valid_to_ms") := {
  vf <- vapply(edge_id, function(eid) {
    e <- tryCatch(client$get_edge(eid), error = function(e) NULL)
    if (!is.null(e) && !is.null(e$valid_from)) e$valid_from else NA_real_
  }, numeric(1))
  vt <- vapply(edge_id, function(eid) {
    e <- tryCatch(client$get_edge(eid), error = function(e) NULL)
    if (!is.null(e) && !is.null(e$valid_to)) e$valid_to else NA_real_
  }, numeric(1))
  list(vf, vt)
}]
# Build lookup and add derived columns.
ip_lookup <- setNames(nodes_dt$ip, as.character(nodes_dt$node_id))

edges_dt[, service_port_int := suppressWarnings(as.integer(service_port))]
edges_dt[, timestamp := as.POSIXct(valid_from_ms / 1000,
                                    origin = "1970-01-01")]
edges_dt[, timestamp_end := as.POSIXct(valid_to_ms / 1000,
                                        origin = "1970-01-01")]
edges_dt[, flow_duration_s := (valid_to_ms - valid_from_ms) / 1000]

# Use package function for service name lookup on missing names
edges_dt[is.na(service_name) & !is.na(service_port_int),
         service_name := port_service_name(service_port_int)]
total_packets <- sum(edges_dt$flow_count, na.rm = TRUE)
summary_stats <- data.frame(
  Metric = c("Unique IP addresses",
             "Total flows (edges)",
             "Total packets (aggregated)",
             "Protocols observed",
             "Time span (seconds)",
             "Total bytes captured",
             "Avg packets per flow"),
  Value = c(
    nrow(nodes_dt),
    nrow(edges_dt),
    total_packets,
    length(unique(edges_dt$protocol)),
    if (nrow(edges_dt) > 0)
      round(as.numeric(difftime(max(edges_dt$timestamp, na.rm = TRUE),
                                min(edges_dt$timestamp, na.rm = TRUE),
                                units = "secs")), 1)
    else 0,
    sum(edges_dt$total_bytes, na.rm = TRUE),
    if (nrow(edges_dt) > 0) round(total_packets / nrow(edges_dt), 1) else 0
  )
)
knitr::kable(summary_stats, caption = "Capture summary")

2 – Protocol Distribution

proto_counts <- edges_dt[, .(
  flows   = .N,
  packets = sum(flow_count, na.rm = TRUE)
), by = protocol][order(-packets)]

ggplot(proto_counts, aes(x = reorder(protocol, packets),
                         y = packets, fill = protocol)) +
  geom_col(show.legend = FALSE) +
  coord_flip() +
  labs(title = "Packets by Protocol (aggregated from flows)",
       x = "Protocol", y = "Packet count") +
  theme_minimal(base_size = 14)
proto_bytes <- edges_dt[, .(
  total_bytes  = sum(total_bytes, na.rm = TRUE),
  total_pkts   = sum(flow_count, na.rm = TRUE),
  avg_pkt_size = round(sum(total_bytes, na.rm = TRUE) /
                         sum(flow_count, na.rm = TRUE)),
  flow_count   = .N
), by = protocol][order(-total_bytes)]

knitr::kable(proto_bytes, caption = "Traffic volume by protocol",
             format.args = list(big.mark = ","))

3 – Top Talkers

By total packets sent

top_senders <- edges_dt[, .(packets_sent = sum(flow_count, na.rm = TRUE)),
                        by = src_ip][order(-packets_sent)][1:min(.N, 20)]

ggplot(top_senders, aes(x = reorder(src_ip, packets_sent), y = packets_sent)) +
  geom_col(fill = "#3498db") +
  coord_flip() +
  labs(title = "Top 20 Source IPs by Packet Count",
       x = NULL, y = "Packets sent") +
  theme_minimal(base_size = 13)

By total bytes sent

top_bytes <- edges_dt[, .(total_bytes = sum(total_bytes, na.rm = TRUE)),
                      by = src_ip][order(-total_bytes)][1:min(.N, 20)]

ggplot(top_bytes, aes(x = reorder(src_ip, total_bytes), y = total_bytes)) +
  geom_col(fill = "#2ecc71") +
  coord_flip() +
  labs(title = "Top 20 Source IPs by Bytes Sent",
       x = NULL, y = "Bytes") +
  scale_y_continuous(labels = label_bytes()) +
  theme_minimal(base_size = 13)

Busiest conversations (IP pairs)

pair_traffic <- edges_dt[, .(
  flows   = .N,
  packets = sum(flow_count, na.rm = TRUE),
  bytes   = sum(total_bytes, na.rm = TRUE)
), by = .(src_ip, dst_ip)][order(-packets)]

knitr::kable(head(pair_traffic, 20),
             caption = "Top 20 IP-pair conversations by packet count",
             format.args = list(big.mark = ","))

4 – Port Analysis

Service Port Distribution

svc_counts <- edges_dt[!is.na(service_port_int), .(
  flows   = .N,
  packets = sum(flow_count, na.rm = TRUE),
  bytes   = sum(total_bytes, na.rm = TRUE)
), by = .(service_port_int, service_name)][order(-packets)]

svc_counts[, label := fifelse(
  !is.na(service_name) & service_name != "",
  paste0(service_port_int, " (", service_name, ")"),
  as.character(service_port_int)
)]

top_svcs <- svc_counts[1:min(.N, 20)]

ggplot(top_svcs, aes(x = reorder(label, packets), y = packets)) +
  geom_col(fill = "#9b59b6") +
  coord_flip() +
  labs(title = "Top 20 Service Ports by Packet Count",
       x = NULL, y = "Packet count (aggregated)") +
  theme_minimal(base_size = 13)
ggplot(top_svcs, aes(x = reorder(label, bytes), y = bytes)) +
  geom_col(fill = "#e67e22") +
  coord_flip() +
  labs(title = "Top 20 Service Ports by Total Bytes",
       x = NULL, y = "Total bytes") +
  scale_y_continuous(labels = label_bytes()) +
  theme_minimal(base_size = 13)

Port-protocol cross-tabulation

pp <- edges_dt[!is.na(service_port_int), .(
  packets = sum(flow_count, na.rm = TRUE)
), by = .(protocol, service_port_int)][order(-packets)]
pp_top <- pp[service_port_int %in% svc_counts$service_port_int[1:15]]

if (nrow(pp_top) > 0) {
  ggplot(pp_top, aes(x = protocol,
                     y = factor(service_port_int),
                     fill = log10(packets + 1))) +
    geom_tile(color = "white") +
    scale_fill_viridis_c(name = "log10(packets)") +
    labs(title = "Protocol vs Service Port",
         x = "Protocol", y = "Service port") +
    theme_minimal(base_size = 13)
}

5 – Temporal Patterns

if (sum(!is.na(edges_dt$timestamp)) > 0) {
  ts_dt <- edges_dt[!is.na(timestamp)]
  # Bin flow arrivals into 1-second intervals, weighting by flow_count
  ts_dt[, second := as.POSIXct(floor(as.numeric(timestamp)),
                                origin = "1970-01-01")]
  ts_agg <- ts_dt[, .(packets = sum(flow_count, na.rm = TRUE),
                       bytes = sum(total_bytes, na.rm = TRUE),
                       flows = .N),
                  by = second][order(second)]

  ggplot(ts_agg, aes(x = second, y = packets)) +
    geom_line(color = "#2c3e50", linewidth = 0.5) +
    geom_smooth(method = "loess", se = TRUE, color = "#e74c3c", span = 0.3) +
    labs(title = "Packet Rate Over Time (from flow aggregates)",
         x = "Time", y = "Packets per second") +
    theme_minimal(base_size = 13)
}
if (exists("ts_agg") && nrow(ts_agg) > 0) {
  ggplot(ts_agg, aes(x = second, y = flows)) +
    geom_line(color = "#1abc9c", linewidth = 0.5) +
    geom_smooth(method = "loess", se = TRUE, color = "#8e44ad", span = 0.3) +
    labs(title = "Flow Arrival Rate Over Time",
         x = "Time", y = "New flows per second") +
    theme_minimal(base_size = 13)
}
if (sum(!is.na(edges_dt$timestamp)) > 0) {
  ts_proto <- ts_dt[, .(packets = sum(flow_count, na.rm = TRUE)),
                    by = .(second, protocol)][order(second)]

  ggplot(ts_proto, aes(x = second, y = packets, fill = protocol)) +
    geom_area(alpha = 0.7, position = "stack") +
    labs(title = "Traffic Volume by Protocol Over Time",
         x = "Time", y = "Packets per second", fill = "Protocol") +
    theme_minimal(base_size = 13)
}
if (sum(!is.na(edges_dt$flow_duration_s)) > 0) {
  ggplot(edges_dt[!is.na(flow_duration_s) & flow_duration_s > 0],
         aes(x = flow_duration_s)) +
    geom_histogram(fill = "#2c3e50", color = "white", bins = 50) +
    scale_x_log10() +
    labs(title = "Flow Duration Distribution",
         x = "Duration (seconds, log scale)", y = "Number of flows") +
    theme_minimal(base_size = 13)
}

6 – Network Graph Analysis

Build an igraph network where edge weights are packet counts between IP pairs.

# Aggregate to one edge per (src, dst, protocol) tuple, summing flow counts
agg <- edges_dt[, .(
  weight      = sum(flow_count, na.rm = TRUE),
  total_bytes = sum(total_bytes, na.rm = TRUE),
  num_flows   = .N
), by = .(src_ip, dst_ip, protocol)]

# Use only IPs that appear in edges (some nodes may be isolates)
all_ips <- unique(c(edges_dt$src_ip, edges_dt$dst_ip, nodes_dt$ip))
g <- graph_from_data_frame(
  d        = agg[, .(from = src_ip, to = dst_ip,
                      weight = weight, protocol = protocol,
                      total_bytes = total_bytes,
                      num_flows = num_flows)],
  directed = TRUE,
  vertices = data.frame(name = all_ips, stringsAsFactors = FALSE)
)

cat(sprintf("Graph: %d vertices, %d aggregated edges\n",
            vcount(g), ecount(g)))

Degree distribution

deg_in  <- degree(g, mode = "in")
deg_out <- degree(g, mode = "out")
deg_all <- degree(g, mode = "all")

deg_df <- data.frame(
  ip     = names(deg_all),
  degree_in  = deg_in,
  degree_out = deg_out,
  degree_total = deg_all,
  stringsAsFactors = FALSE
)

ggplot(data.frame(degree = deg_all), aes(x = degree)) +
  geom_histogram(fill = "#1abc9c", color = "white", bins = 30) +
  labs(title = "Degree Distribution (all directions)",
       x = "Degree", y = "Number of IPs") +
  theme_minimal(base_size = 13)
# Check for power-law / heavy tail
deg_tab <- as.data.frame(table(degree = deg_all))
deg_tab$degree <- as.integer(as.character(deg_tab$degree))
deg_tab <- deg_tab[deg_tab$degree > 0, ]

if (nrow(deg_tab) > 3) {
  ggplot(deg_tab, aes(x = degree, y = Freq)) +
    geom_point(color = "#e67e22", size = 2) +
    scale_x_log10() + scale_y_log10() +
    labs(title = "Degree Distribution (log-log scale)",
         subtitle = "A straight line suggests scale-free / power-law structure",
         x = "Degree (log)", y = "Count (log)") +
    theme_minimal(base_size = 13)
}

Centrality measures

# Compute centralities on the simplified (no multi-edge) graph
gs <- simplify(g, edge.attr.comb = list(weight = "sum",
                                         total_bytes = "sum",
                                         num_flows = "sum",
                                         protocol = "first"))

centrality_dt <- data.table(
  ip        = V(gs)$name,
  degree    = degree(gs, mode = "all"),
  in_degree = degree(gs, mode = "in"),
  out_degree = degree(gs, mode = "out"),
  strength  = strength(gs, mode = "all"),
  betweenness = round(betweenness(gs, directed = TRUE, normalized = TRUE), 6),
  pagerank  = round(page_rank(gs, directed = TRUE)$vector, 6)
)
setorder(centrality_dt, -pagerank)

knitr::kable(head(centrality_dt, 20),
             caption = "Top 20 IPs by PageRank")

Community detection

g_undir <- as.undirected(gs, mode = "collapse",
                         edge.attr.comb = list(weight = "sum",
                                               total_bytes = "sum",
                                               num_flows = "sum",
                                               protocol = "first"))
if (vcount(g_undir) >= 3) {
  comm <- cluster_louvain(g_undir, weights = E(g_undir)$weight)
  cat(sprintf("Louvain detected %d communities (modularity = %.3f)\n",
              length(comm), modularity(comm)))

  comm_dt <- data.table(
    ip = V(g_undir)$name,
    community = membership(comm)
  )
  comm_sizes <- comm_dt[, .N, by = community][order(-N)]
  setnames(comm_sizes, "N", "members")

  knitr::kable(comm_sizes, caption = "Community sizes")

  # Store membership for visualisation
  V(g_undir)$community <- membership(comm)
}

7 – Anomaly Indicators

Flag IPs that exhibit unusual behaviour: high fan-out to many destinations (potential scanning), connections to many distinct service ports (potential reconnaissance), or asymmetric traffic ratios.

anomaly_dt <- edges_dt[, .(
  unique_dst_ips   = uniqueN(dst_ip),
  unique_svc_ports = uniqueN(service_port_int, na.rm = TRUE),
  packets_sent     = sum(flow_count, na.rm = TRUE),
  bytes_sent       = sum(total_bytes, na.rm = TRUE),
  flows_out        = .N
), by = src_ip]

# Incoming stats
incoming <- edges_dt[, .(
  packets_received = sum(flow_count, na.rm = TRUE),
  bytes_received   = sum(total_bytes, na.rm = TRUE),
  flows_in         = .N
), by = .(dst_ip)]

anomaly_dt <- merge(anomaly_dt, incoming,
                    by.x = "src_ip", by.y = "dst_ip", all.x = TRUE)
anomaly_dt[is.na(packets_received), packets_received := 0L]
anomaly_dt[is.na(bytes_received), bytes_received := 0L]
anomaly_dt[is.na(flows_in), flows_in := 0L]

# Asymmetry ratio: > 1 means more sending than receiving
anomaly_dt[, send_recv_ratio := fifelse(
  packets_received > 0,
  round(packets_sent / packets_received, 2),
  Inf
)]

Potential scanners (high destination fan-out)

scanner_thresh <- quantile(anomaly_dt$unique_dst_ips, 0.9, na.rm = TRUE)
scanners <- anomaly_dt[unique_dst_ips >= max(scanner_thresh, 3)][order(-unique_dst_ips)]

if (nrow(scanners) > 0) {
  ggplot(scanners, aes(x = reorder(src_ip, unique_dst_ips),
                       y = unique_dst_ips)) +
    geom_col(fill = "#e74c3c") +
    coord_flip() +
    labs(title = "IPs with High Destination Fan-Out",
         subtitle = "Potential network scanning behaviour",
         x = NULL, y = "Unique destination IPs") +
    theme_minimal(base_size = 13)
}

knitr::kable(head(scanners, 15),
             caption = "IPs contacting the most unique destinations",
             format.args = list(big.mark = ","))

Port reconnaissance (many distinct service ports)

recon_thresh <- quantile(anomaly_dt$unique_svc_ports, 0.9, na.rm = TRUE)
recon <- anomaly_dt[unique_svc_ports >= max(recon_thresh, 5)][order(-unique_svc_ports)]

if (nrow(recon) > 0) {
  ggplot(recon, aes(x = reorder(src_ip, unique_svc_ports),
                    y = unique_svc_ports)) +
    geom_col(fill = "#f39c12") +
    coord_flip() +
    labs(title = "IPs Targeting Many Distinct Service Ports",
         subtitle = "Potential port reconnaissance",
         x = NULL, y = "Unique service ports") +
    theme_minimal(base_size = 13)
}

Send/receive asymmetry

asym <- anomaly_dt[is.finite(send_recv_ratio)][order(-send_recv_ratio)]

if (nrow(asym) > 0) {
  ggplot(asym, aes(x = packets_sent, y = packets_received,
                   color = send_recv_ratio)) +
    geom_point(size = 3, alpha = 0.8) +
    geom_abline(slope = 1, intercept = 0, linetype = "dashed",
                color = "grey50") +
    scale_color_viridis_c(name = "Send/Recv\nratio", trans = "log1p") +
    labs(title = "Traffic Symmetry per IP",
         subtitle = "Points above the dashed line receive more than they send",
         x = "Packets sent", y = "Packets received") +
    theme_minimal(base_size = 13)
}

8 – Interactive Network Map

# Prepare visNetwork data from the simplified igraph
vis_nodes <- data.frame(
  id    = V(gs)$name,
  label = V(gs)$name,
  stringsAsFactors = FALSE
)

# Size nodes by total strength (packets in + out)
node_strength <- strength(gs, mode = "all")
vis_nodes$value <- as.numeric(node_strength[vis_nodes$id])

# Colour by community if available
if (!is.null(V(g_undir)$community)) {
  comm_map <- setNames(V(g_undir)$community, V(g_undir)$name)
  vis_nodes$group <- as.character(comm_map[vis_nodes$id])
}

vis_edges <- data.frame(
  from   = ends(gs, E(gs))[, 1],
  to     = ends(gs, E(gs))[, 2],
  value  = E(gs)$weight,
  title  = paste0(E(gs)$protocol, " — ",
                  format(E(gs)$weight, big.mark = ","), " packets, ",
                  format(E(gs)$total_bytes, big.mark = ","), " bytes"),
  arrows = "to",
  stringsAsFactors = FALSE
)

visNetwork(vis_nodes, vis_edges,
           main = "Captured IP Traffic Graph",
           submain = paste(nrow(vis_nodes), "hosts,",
                           nrow(vis_edges), "aggregated connections")) %>%
  visOptions(highlightNearest = list(enabled = TRUE, degree = 1),
             nodesIdSelection = TRUE) %>%
  visPhysics(solver = "forceAtlas2Based",
             forceAtlas2Based = list(gravitationalConstant = -50)) %>%
  visInteraction(navigationButtons = TRUE)

9 – Payload Analysis

Examine the distribution of traffic volumes across flows to identify traffic profiles (e.g., small DNS lookups vs large data transfers).

edges_dt[, avg_pkt_size := fifelse(
  flow_count > 0,
  total_bytes / flow_count,
  NA_real_
)]
ggplot(edges_dt[!is.na(total_bytes)],
       aes(x = total_bytes)) +
  geom_histogram(fill = "#2c3e50", color = "white", bins = 50) +
  scale_x_log10(labels = label_bytes()) +
  labs(title = "Flow Size Distribution",
       x = "Total bytes per flow (log scale)", y = "Number of flows") +
  theme_minimal(base_size = 13)
if (edges_dt[, uniqueN(protocol)] > 1) {
  ggplot(edges_dt[!is.na(avg_pkt_size)],
         aes(x = protocol, y = avg_pkt_size, fill = protocol)) +
    geom_violin(show.legend = FALSE, alpha = 0.7) +
    geom_boxplot(width = 0.15, outlier.size = 0.5, show.legend = FALSE) +
    labs(title = "Average Packet Size by Protocol",
         subtitle = "total_bytes / flow_count for each flow",
         x = "Protocol", y = "Average packet size (bytes)") +
    theme_minimal(base_size = 13)
}
ggplot(edges_dt[flow_count > 0],
       aes(x = flow_count)) +
  geom_histogram(fill = "#3498db", color = "white", bins = 50) +
  scale_x_log10() +
  labs(title = "Packets per Flow Distribution",
       x = "Packets per flow (log scale)", y = "Number of flows") +
  theme_minimal(base_size = 13)

10 – AstraeaDB Graph Traversal Examples

Demonstrate using AstraeaDB’s built-in graph algorithms on the captured data.

# BFS from the highest-PageRank node
top_ip     <- centrality_dt$ip[1]
top_ip_id  <- nodes_dt[ip == top_ip, node_id]
if (length(top_ip_id) == 1) {
  bfs_result <- client$bfs(top_ip_id, max_depth = 2L)
  cat(sprintf(
    "BFS from %s (node %d): reached %d nodes within 2 hops.\n",
    top_ip, top_ip_id, length(bfs_result)
  ))
}
# Shortest path between the two highest-PageRank nodes
if (nrow(centrality_dt) >= 2) {
  id_a <- nodes_dt[ip == centrality_dt$ip[1], node_id]
  id_b <- nodes_dt[ip == centrality_dt$ip[2], node_id]
  if (length(id_a) == 1 && length(id_b) == 1) {
    sp <- tryCatch(
      client$shortest_path(id_a, id_b, weighted = FALSE),
      error = function(e) NULL
    )
    if (!is.null(sp) && length(sp$path) > 0) {
      path_ips <- sapply(sp$path, function(nid) ip_lookup[as.character(nid)])
      cat(sprintf("Shortest path (%d hops): %s\n",
                  length(sp$path) - 1L,
                  paste(path_ips, collapse = " -> ")))
    } else {
      cat("No path found between the top two IPs.\n")
    }
  }
}

Analysis generated on 2026-08-19 22:12:25.44555 from AstraeaDB captured traffic.

These binaries (installable software) and packages are in development.
They may not be fully stable and should be used with caution. We make no claims about them.
Health stats visible at Monitor.