From 26ab808730df0ec8b29de3b67032b380a2e53211 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Smolarek?=
 <34063647+Razz4780@users.noreply.github.com>
Date: Thu, 2 Jan 2025 15:08:27 +0100
Subject: [PATCH] Fix port mirror rejection in intproxy (#2998)

* Fixed handling rejection in intproxy

* Changelog

* fmt
---
 .../+mirrord-policy-rejection.fixed.md        |  1 +
 .../src/proxies/incoming/subscriptions.rs     | 26 ++++++++++++-------
 2 files changed, 18 insertions(+), 9 deletions(-)
 create mode 100644 changelog.d/+mirrord-policy-rejection.fixed.md

diff --git a/changelog.d/+mirrord-policy-rejection.fixed.md b/changelog.d/+mirrord-policy-rejection.fixed.md
new file mode 100644
index 00000000000..ad7415c5411
--- /dev/null
+++ b/changelog.d/+mirrord-policy-rejection.fixed.md
@@ -0,0 +1 @@
+Fixed a bug where port mirroring block (due to active mirrord policies) would terminate the mirrord session.
diff --git a/mirrord/intproxy/src/proxies/incoming/subscriptions.rs b/mirrord/intproxy/src/proxies/incoming/subscriptions.rs
index 8439e79d8e1..7731707d8c8 100644
--- a/mirrord/intproxy/src/proxies/incoming/subscriptions.rs
+++ b/mirrord/intproxy/src/proxies/incoming/subscriptions.rs
@@ -242,6 +242,7 @@ impl SubscriptionsManager {
 
                 Ok(subscription.confirm())
             }
+
             Err(ResponseError::PortAlreadyStolen(port)) => {
                 let Some(subscription) = self.subscriptions.remove(&port) else {
                     return Ok(vec![]);
@@ -255,23 +256,30 @@ impl SubscriptionsManager {
                     }
                 }
             }
+
             Err(
-                ref response_err @ ResponseError::Forbidden {
-                    blocked_action: BlockedAction::Steal(ref steal_type),
-                    ..
+                ref response_error @ ResponseError::Forbidden {
+                    ref blocked_action, ..
                 },
             ) => {
-                tracing::warn!("Port subscribe blocked by policy: {response_err}");
-                let Some(subscription) = self.subscriptions.remove(&steal_type.get_port()) else {
+                tracing::warn!(%response_error, "Port subscribe blocked by policy");
+
+                let port = match blocked_action {
+                    BlockedAction::Steal(steal_type) => steal_type.get_port(),
+                    BlockedAction::Mirror(port) => *port,
+                };
+                let Some(subscription) = self.subscriptions.remove(&port) else {
                     return Ok(vec![]);
                 };
+
                 subscription
-                    .reject(response_err.clone())
-                    .map_err(|sub|{
-                        tracing::error!("Subscription {sub:?} was confirmed before, then requested again and blocked by a policy.");
-                        IncomingProxyError::SubscriptionFailed(response_err.clone())
+                    .reject(response_error.clone())
+                    .map_err(|subscription|{
+                        tracing::error!(?subscription, "Subscription was confirmed before, then requested again and blocked by a policy.");
+                        IncomingProxyError::SubscriptionFailed(response_error.clone())
                     })
             }
+
             Err(err) => Err(IncomingProxyError::SubscriptionFailed(err)),
         }
     }