feat: add include_rules and include_flow_context query parameters to workload insights endpoint for conditional rule/policy matching
Add include_rules boolean parameter to conditionally fetch active firewall rules (defaults to false), add include_flow_context boolean parameter to conditionally compute matching firewall rules and policies for each flow (defaults to false), rename traffic parameter to traffic_mode with alias for backward compatibility, update flow decision logic to return "observed" when
This commit is contained in:
@@ -1517,7 +1517,9 @@ async def workload_insights(
|
||||
workload_id: str,
|
||||
_: CurrentUser,
|
||||
db: Session = Depends(get_db),
|
||||
traffic: str = Query(default="summary", pattern="^(summary|full)$"),
|
||||
traffic_mode: str = Query(default="summary", alias="traffic", pattern="^(summary|full)$"),
|
||||
include_rules: bool = False,
|
||||
include_flow_context: bool = False,
|
||||
) -> WorkloadInsight:
|
||||
workload = db.get(Workload, workload_id)
|
||||
if not workload:
|
||||
@@ -1536,8 +1538,8 @@ async def workload_insights(
|
||||
if address.workload_id:
|
||||
workload_ips_by_id.setdefault(address.workload_id, set()).add(address.address)
|
||||
known_subnets = db.scalars(select(Subnet).order_by(Subnet.cidr)).all()
|
||||
active_firewall_rules = await active_firewall_rules_for_workload(db, workload)
|
||||
traffic = []
|
||||
active_firewall_rules = await active_firewall_rules_for_workload(db, workload) if include_rules else []
|
||||
traffic_rows = []
|
||||
if workload_ips:
|
||||
workload_ip_set = set(workload_ips)
|
||||
flow_query = (
|
||||
@@ -1545,25 +1547,28 @@ async def workload_insights(
|
||||
.where((TrafficFlow.source_ip.in_(workload_ips)) | (TrafficFlow.destination_ip.in_(workload_ips)))
|
||||
.order_by((TrafficFlow.state == "blocked").desc(), TrafficFlow.updated_at.desc())
|
||||
)
|
||||
if traffic == "summary":
|
||||
if traffic_mode == "summary":
|
||||
flow_query = flow_query.limit(250)
|
||||
flows = db.scalars(flow_query).all()
|
||||
for flow in flows:
|
||||
source_owner = ip_owners.get(flow.source_ip)
|
||||
destination_owner = ip_owners.get(flow.destination_ip)
|
||||
matching_firewall_rules = [
|
||||
firewall_rule_flow_payload(rule)
|
||||
for rule in active_firewall_rules
|
||||
if "error" not in rule and firewall_rule_matches_flow(rule, flow, workload_ip_set)
|
||||
]
|
||||
matching_policies = [
|
||||
policy_flow_payload(policy)
|
||||
for policy in policies
|
||||
if policy_matches_flow(db, policy, flow, ip_owners, workload_ips_by_id, workload.cluster_id)
|
||||
]
|
||||
matching_firewall_rules = []
|
||||
matching_policies = []
|
||||
if include_flow_context:
|
||||
matching_firewall_rules = [
|
||||
firewall_rule_flow_payload(rule)
|
||||
for rule in active_firewall_rules
|
||||
if "error" not in rule and firewall_rule_matches_flow(rule, flow, workload_ip_set)
|
||||
]
|
||||
matching_policies = [
|
||||
policy_flow_payload(policy)
|
||||
for policy in policies
|
||||
if policy_matches_flow(db, policy, flow, ip_owners, workload_ips_by_id, workload.cluster_id)
|
||||
]
|
||||
raw_decision = raw_flow_decision(flow)
|
||||
raw_payload = flow.raw if isinstance(flow.raw, dict) else {}
|
||||
traffic.append(
|
||||
traffic_rows.append(
|
||||
{
|
||||
"source": flow_endpoint_label(source_owner, known_subnets, flow.source_ip),
|
||||
"destination": flow_endpoint_label(destination_owner, known_subnets, flow.destination_ip),
|
||||
@@ -1577,7 +1582,7 @@ async def workload_insights(
|
||||
"bytes": flow.bytes,
|
||||
"packets": flow.packets,
|
||||
"state": flow.state,
|
||||
"decision": raw_decision or flow_policy_decision(matching_firewall_rules, matching_policies),
|
||||
"decision": raw_decision or (flow_policy_decision(matching_firewall_rules, matching_policies) if include_flow_context else "observed"),
|
||||
"collector": raw_payload.get("collector"),
|
||||
"matching_firewall_rules": matching_firewall_rules,
|
||||
"matching_audit_policies": [
|
||||
@@ -1588,13 +1593,13 @@ async def workload_insights(
|
||||
"ip_addresses": [flow.source_ip, flow.destination_ip],
|
||||
}
|
||||
)
|
||||
if not traffic:
|
||||
if not traffic_rows:
|
||||
agent = db.get(NodeAgent, workload.node_id)
|
||||
payload = agent.last_payload if agent and isinstance(agent.last_payload, dict) else {}
|
||||
for item in payload.get("interface_traffic", []):
|
||||
if not isinstance(item, dict) or str(item.get("vmid")) != str(workload.external_id):
|
||||
continue
|
||||
traffic.append(
|
||||
traffic_rows.append(
|
||||
{
|
||||
"source": workload.name,
|
||||
"destination": "network",
|
||||
@@ -1628,7 +1633,7 @@ async def workload_insights(
|
||||
return WorkloadInsight(
|
||||
workload=workload,
|
||||
assigned_ips=[ip_address_payload(db, address) for address in assigned_ips],
|
||||
traffic=traffic,
|
||||
traffic=traffic_rows,
|
||||
active_firewall_rules=active_firewall_rules,
|
||||
matching_policies=policies,
|
||||
effective_decision=decision,
|
||||
|
||||
@@ -453,7 +453,7 @@ export function Workloads() {
|
||||
const selected = selectedId || workloads.data?.[0]?.id || "";
|
||||
const insight = useQuery({
|
||||
queryKey: ["workload-insight", selected, "summary"],
|
||||
queryFn: () => api<WorkloadInsight>(`/vms/${selected}/insights?traffic=summary`),
|
||||
queryFn: () => api<WorkloadInsight>(`/vms/${selected}/insights?traffic=summary&include_rules=false&include_flow_context=false`),
|
||||
enabled: Boolean(selected),
|
||||
});
|
||||
const traffic = useMemo(() => summarizeTraffic(insight.data?.traffic ?? []), [insight.data?.traffic]);
|
||||
@@ -540,8 +540,8 @@ export function Workloads() {
|
||||
export function WorkloadDetail() {
|
||||
const { workloadId } = useParams();
|
||||
const insight = useQuery({
|
||||
queryKey: ["workload-insight", workloadId, "summary"],
|
||||
queryFn: () => api<WorkloadInsight>(`/vms/${workloadId}/insights?traffic=summary`),
|
||||
queryKey: ["workload-insight", workloadId, "summary", "rules"],
|
||||
queryFn: () => api<WorkloadInsight>(`/vms/${workloadId}/insights?traffic=summary&include_rules=true&include_flow_context=true`),
|
||||
enabled: Boolean(workloadId),
|
||||
});
|
||||
const traffic = useMemo(() => summarizeTraffic(insight.data?.traffic ?? []), [insight.data?.traffic]);
|
||||
@@ -639,7 +639,7 @@ export function WorkloadFlows() {
|
||||
const [page, setPage] = useState(1);
|
||||
const insight = useQuery({
|
||||
queryKey: ["workload-insight", workloadId, "full"],
|
||||
queryFn: () => api<WorkloadInsight>(`/vms/${workloadId}/insights?traffic=full`),
|
||||
queryFn: () => api<WorkloadInsight>(`/vms/${workloadId}/insights?traffic=full&include_rules=false&include_flow_context=false`),
|
||||
enabled: Boolean(workloadId),
|
||||
});
|
||||
const traffic = useMemo(() => summarizeTraffic(insight.data?.traffic ?? []), [insight.data?.traffic]);
|
||||
|
||||
Reference in New Issue
Block a user