l7policy: guard missing upstream addresses in response filter

cilium/l7policy.cc

@@ -343,10 +343,13 @@ Http::FilterHeadersStatus AccessFilter::encodeHeaders(Http::ResponseHeaderMap& h
 
     // check if upstream and downstream connections have the same source and destination
     // addresses, respectively (note: do not compare pointers!).
-    if (*upstream_info.upstreamRemoteAddress() ==
-            *stream_info.downstreamAddressProvider().localAddress() &&
-        *upstream_info.upstreamLocalAddress() ==
-            *stream_info.downstreamAddressProvider().remoteAddress()) {
+    const auto upstream_remote_address = upstream_info.upstreamRemoteAddress();
+    const auto upstream_local_address = upstream_info.upstreamLocalAddress();
+    const auto downstream_local_address = stream_info.downstreamAddressProvider().localAddress();
+    const auto downstream_remote_address = stream_info.downstreamAddressProvider().remoteAddress();
+    if (upstream_remote_address && upstream_local_address && downstream_local_address &&
+        downstream_remote_address && *upstream_remote_address == *downstream_local_address &&
+        *upstream_local_address == *downstream_remote_address) {
       ENVOY_CONN_LOG(debug,
                      "cilium.l7policy: Upstream connection with same 5-tuple closed, passing "
                      "connection close to downstream response",

tests/BUILD

@@ -208,6 +208,23 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "l7policy_test",
+    srcs = ["l7policy_test.cc"],
+    repository = "@envoy",
+    deps = [
+        "//cilium:l7policy_lib",
+        "@envoy//source/common/network:address_lib",
+        "@envoy//source/common/stats:isolated_store_lib",
+        "@envoy//source/common/stream_info:stream_info_lib",
+        "@envoy//test/mocks/http:http_mocks",
+        "@envoy//test/mocks/network:connection_mocks",
+        "@envoy//test/mocks/stream_info:stream_info_mocks",
+        "@envoy//test/test_common:simulated_time_system_lib",
+        "@envoy//test/test_common:utility_lib",
+    ],
+)
+
 envoy_cc_test(
     name = "cilium_tcp_integration_test",
     srcs = ["cilium_tcp_integration_test.cc"],

tests/l7policy_test.cc

@@ -0,0 +1,104 @@
+#include <memory>
+
+#include "envoy/common/optref.h"
+#include "envoy/http/filter.h"
+#include "envoy/http/protocol.h"
+#include "envoy/network/connection.h"
+#include "envoy/stream_info/stream_info.h"
+#include "envoy/upstream/host_description.h"
+
+#include "source/common/network/address_impl.h"
+#include "source/common/stats/isolated_store_impl.h"
+#include "source/common/stream_info/stream_info_impl.h"
+
+#include "test/mocks/http/mocks.h"
+#include "test/mocks/network/connection.h"
+#include "test/test_common/simulated_time_system.h"
+#include "test/test_common/utility.h"
+
+#include "cilium/accesslog.h"
+#include "cilium/l7policy.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace Cilium {
+namespace {
+
+class TestStreamDecoderFilterCallbacks : public Http::MockStreamDecoderFilterCallbacks {
+public:
+  bool iterateUpstreamCallbacks(Upstream::HostDescriptionConstSharedPtr,
+                                StreamInfo::StreamInfo&) override {
+    return true;
+  }
+};
+
+class L7PolicyResponseTest : public testing::Test {
+protected:
+  L7PolicyResponseTest()
+      : config_(std::make_shared<Config>("", "", time_system_, *stats_.rootScope(), false)),
+        filter_(config_) {
+    ON_CALL(callbacks_, connection())
+        .WillByDefault(testing::Return(OptRef<const Network::Connection>{connection_}));
+    callbacks_.stream_info_.protocol_ = Http::Protocol::Http11;
+    callbacks_.stream_info_.upstream_info_ = upstream_info_;
+    filter_.setDecoderFilterCallbacks(callbacks_);
+  }
+
+  void expectLocalReplyWithoutDraining() {
+    EXPECT_CALL(callbacks_.stream_info_, setShouldDrainConnectionUponCompletion(true)).Times(0);
+    EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_.encodeHeaders(headers_, true));
+    EXPECT_EQ("503", headers_.getStatusValue());
+    const auto* log_entry =
+        callbacks_.stream_info_.filter_state_->getDataReadOnly<AccessLog::Entry>(AccessLogKey);
+    ASSERT_NE(nullptr, log_entry);
+    EXPECT_EQ(503, log_entry->entry_.http().status());
+  }
+
+  Event::SimulatedTimeSystem time_system_;
+  Stats::IsolatedStoreImpl stats_;
+  testing::NiceMock<Network::MockConnection> connection_;
+  testing::NiceMock<TestStreamDecoderFilterCallbacks> callbacks_;
+  std::shared_ptr<StreamInfo::UpstreamInfoImpl> upstream_info_ =
+      std::make_shared<StreamInfo::UpstreamInfoImpl>();
+  ConfigSharedPtr config_;
+  AccessFilter filter_;
+  Http::TestResponseHeaderMapImpl headers_{{":status", "503"}, {"connection", "close"}};
+};
+
+TEST_F(L7PolicyResponseTest, LocalReplyWithoutUpstreamSocketAddresses) {
+  // A local reply can have upstream info even though no upstream socket was established.
+  expectLocalReplyWithoutDraining();
+}
+
+TEST_F(L7PolicyResponseTest, LocalReplyWithoutUpstreamLocalAddress) {
+  upstream_info_->setUpstreamRemoteAddress(
+      callbacks_.stream_info_.downstreamAddressProvider().localAddress());
+  expectLocalReplyWithoutDraining();
+}
+
+TEST_F(L7PolicyResponseTest, LocalReplyWithoutUpstreamRemoteAddress) {
+  upstream_info_->setUpstreamLocalAddress(
+      callbacks_.stream_info_.downstreamAddressProvider().remoteAddress());
+  expectLocalReplyWithoutDraining();
+}
+
+TEST_F(L7PolicyResponseTest, MatchingAddressesDrainDownstreamConnection) {
+  const auto& downstream = callbacks_.stream_info_.downstreamAddressProvider();
+  upstream_info_->setUpstreamRemoteAddress(downstream.localAddress());
+  upstream_info_->setUpstreamLocalAddress(downstream.remoteAddress());
+  EXPECT_CALL(callbacks_.stream_info_, setShouldDrainConnectionUponCompletion(true));
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_.encodeHeaders(headers_, true));
+}
+
+TEST_F(L7PolicyResponseTest, DifferentAddressesDoNotDrainDownstreamConnection) {
+  upstream_info_->setUpstreamRemoteAddress(
+      std::make_shared<Network::Address::Ipv4Instance>("192.0.2.1", 80));
+  upstream_info_->setUpstreamLocalAddress(
+      callbacks_.stream_info_.downstreamAddressProvider().remoteAddress());
+  expectLocalReplyWithoutDraining();
+}
+
+} // namespace
+} // namespace Cilium
+} // namespace Envoy