diff --git a/Tethering/Android.bp b/Tethering/Android.bp
index d55fef6..70b38a4 100644
--- a/Tethering/Android.bp
+++ b/Tethering/Android.bp
@@ -57,6 +57,7 @@
         "src/**/*.java",
         ":framework-connectivity-shared-srcs",
         ":services-tethering-shared-srcs",
+        ":statslog-connectivity-java-gen",
         ":statslog-tethering-java-gen",
     ],
     static_libs: [
diff --git a/Tethering/common/TetheringLib/Android.bp b/Tethering/common/TetheringLib/Android.bp
index 3ae26cc..2f3307a 100644
--- a/Tethering/common/TetheringLib/Android.bp
+++ b/Tethering/common/TetheringLib/Android.bp
@@ -47,6 +47,7 @@
         "//packages/modules/Connectivity/tests:__subpackages__",
         "//packages/modules/Connectivity/thread/tests:__subpackages__",
         "//packages/modules/IPsec/tests/iketests",
+        "//packages/modules/NetworkStack",
         "//packages/modules/NetworkStack/tests:__subpackages__",
         "//packages/modules/Wifi/service/tests/wifitests",
     ],
diff --git a/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java b/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java
index 5c853f4..89e06da 100644
--- a/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java
+++ b/Tethering/src/com/android/networkstack/tethering/BpfCoordinator.java
@@ -33,10 +33,12 @@
 import static com.android.networkstack.tethering.BpfUtils.DOWNSTREAM;
 import static com.android.networkstack.tethering.BpfUtils.UPSTREAM;
 import static com.android.networkstack.tethering.TetheringConfiguration.DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS;
+import static com.android.networkstack.tethering.TetheringConfiguration.TETHER_ACTIVE_SESSIONS_METRICS;
 import static com.android.networkstack.tethering.UpstreamNetworkState.isVcnInterface;
 import static com.android.networkstack.tethering.util.TetheringUtils.getTetheringJniLibraryName;
 
 import android.app.usage.NetworkStatsManager;
+import android.content.Context;
 import android.net.INetd;
 import android.net.IpPrefix;
 import android.net.LinkProperties;
@@ -65,6 +67,7 @@
 import com.android.net.module.util.BpfDump;
 import com.android.net.module.util.BpfMap;
 import com.android.net.module.util.CollectionUtils;
+import com.android.net.module.util.DeviceConfigUtils;
 import com.android.net.module.util.IBpfMap;
 import com.android.net.module.util.InterfaceParams;
 import com.android.net.module.util.NetworkStackConstants;
@@ -84,6 +87,7 @@
 import com.android.net.module.util.netlink.NetlinkUtils;
 import com.android.networkstack.tethering.apishim.common.BpfCoordinatorShim;
 import com.android.networkstack.tethering.util.TetheringUtils.ForwardedStats;
+import com.android.server.ConnectivityStatsLog;
 
 import java.io.IOException;
 import java.net.Inet4Address;
@@ -148,6 +152,13 @@
 
     @VisibleForTesting
     static final int CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS = 60_000;
+    // The interval is set to 5 minutes to strike a balance between minimizing
+    // the amount of metrics data uploaded and providing sufficient resolution
+    // to track changes in forwarding rules. This choice considers the minimum
+    // push metrics sampling interval of 5 minutes and the 3-minute timeout
+    // for forwarding rules.
+    @VisibleForTesting
+    static final int CONNTRACK_METRICS_UPDATE_INTERVAL_MS = 300_000;
     @VisibleForTesting
     static final int NF_CONNTRACK_TCP_TIMEOUT_ESTABLISHED = 432_000;
     @VisibleForTesting
@@ -314,12 +325,23 @@
         scheduleConntrackTimeoutUpdate();
     };
 
+    private final boolean mSupportActiveSessionsMetrics;
+
+    // Runnable that used by scheduling next refreshing of conntrack metrics sampling.
+    private final Runnable mScheduledConntrackMetricsSampling = () -> {
+        uploadConntrackMetricsSample();
+        scheduleConntrackMetricsSampling();
+    };
+
     // TODO: add BpfMap<TetherDownstream64Key, TetherDownstream64Value> retrieving function.
     @VisibleForTesting
     public abstract static class Dependencies {
         /** Get handler. */
         @NonNull public abstract Handler getHandler();
 
+        /** Get context. */
+        @NonNull public abstract Context getContext();
+
         /** Get netd. */
         @NonNull public abstract INetd getNetd();
 
@@ -472,6 +494,19 @@
                 return null;
             }
         }
+
+        /** Send a TetheringActiveSessionsReported event. */
+        public void sendTetheringActiveSessionsReported(int lastMaxSessionCount) {
+            ConnectivityStatsLog.write(ConnectivityStatsLog.TETHERING_ACTIVE_SESSIONS_REPORTED,
+                    lastMaxSessionCount);
+        }
+
+        /**
+         * @see DeviceConfigUtils#isTetheringFeatureEnabled
+         */
+        public boolean isFeatureEnabled(Context context, String name) {
+            return DeviceConfigUtils.isTetheringFeatureEnabled(context, name);
+        }
     }
 
     @VisibleForTesting
@@ -508,32 +543,57 @@
         if (!mBpfCoordinatorShim.isInitialized()) {
             mLog.e("Bpf shim not initialized");
         }
+
+        // BPF IPv4 forwarding only supports on S+.
+        mSupportActiveSessionsMetrics = mDeps.isAtLeastS()
+                && mDeps.isFeatureEnabled(mDeps.getContext(), TETHER_ACTIVE_SESSIONS_METRICS);
     }
 
     /**
-     * Start BPF tethering offload stats and conntrack timeout polling.
+     * Start BPF tethering offload stats and conntrack polling.
      * Note that this can be only called on handler thread.
      */
-    private void startStatsAndConntrackTimeoutPolling() {
+    private void startStatsAndConntrackPolling() {
         schedulePollingStats();
         scheduleConntrackTimeoutUpdate();
+        if (mSupportActiveSessionsMetrics) {
+            scheduleConntrackMetricsSampling();
+        }
 
         mLog.i("Polling started.");
     }
 
     /**
-     * Stop BPF tethering offload stats and conntrack timeout polling.
+     * Stop BPF tethering offload stats and conntrack polling.
      * The data limit cleanup and the tether stats maps cleanup are not implemented here.
      * These cleanups rely on all IpServers calling #removeIpv6DownstreamRule. After the
      * last rule is removed from the upstream, #removeIpv6DownstreamRule does the cleanup
      * functionality.
      * Note that this can be only called on handler thread.
      */
-    private void stopStatsAndConntrackTimeoutPolling() {
+    private void stopStatsAndConntrackPolling() {
         // Stop scheduled polling conntrack timeout.
         if (mHandler.hasCallbacks(mScheduledConntrackTimeoutUpdate)) {
             mHandler.removeCallbacks(mScheduledConntrackTimeoutUpdate);
         }
+        // Stop scheduled polling conntrack metrics sampling and
+        // clear counters in case there is any counter unsync problem
+        // previously due to possible bpf failures.
+        // Normally this won't happen because all clients are cleared before
+        // reaching here. See IpServer.BaseServingState#exit().
+        if (mSupportActiveSessionsMetrics) {
+            if (mHandler.hasCallbacks(mScheduledConntrackMetricsSampling)) {
+                mHandler.removeCallbacks(mScheduledConntrackMetricsSampling);
+            }
+            final int currentCount = mBpfConntrackEventConsumer.getCurrentConnectionCount();
+            if (currentCount != 0) {
+                Log.wtf(TAG, "Unexpected CurrentConnectionCount: " + currentCount);
+            }
+            // Avoid sending metrics when tethering is about to close.
+            // This leads to a missing final sample before disconnect
+            // but avoids possibly duplicating the last metric in the upload.
+            mBpfConntrackEventConsumer.clearConnectionCounters();
+        }
         // Stop scheduled polling stats and poll the latest stats from BPF maps.
         if (mHandler.hasCallbacks(mScheduledPollingStats)) {
             mHandler.removeCallbacks(mScheduledPollingStats);
@@ -867,7 +927,7 @@
 
         // Start monitoring and polling when the first IpServer is added.
         if (mServedIpServers.isEmpty()) {
-            startStatsAndConntrackTimeoutPolling();
+            startStatsAndConntrackPolling();
             startConntrackMonitoring();
             mIpNeighborMonitor.start();
             mLog.i("Neighbor monitoring started.");
@@ -890,7 +950,7 @@
 
         // Stop monitoring and polling when the last IpServer is removed.
         if (mServedIpServers.isEmpty()) {
-            stopStatsAndConntrackTimeoutPolling();
+            stopStatsAndConntrackPolling();
             stopConntrackMonitoring();
             mIpNeighborMonitor.stop();
             mLog.i("Neighbor monitoring stopped.");
@@ -1031,6 +1091,10 @@
         for (final Tether4Key k : deleteDownstreamRuleKeys) {
             mBpfCoordinatorShim.tetherOffloadRuleRemove(DOWNSTREAM, k);
         }
+        if (mSupportActiveSessionsMetrics) {
+            mBpfConntrackEventConsumer.decreaseCurrentConnectionCount(
+                    deleteUpstreamRuleKeys.size());
+        }
 
         // Cleanup each upstream interface by a set which avoids duplicated work on the same
         // upstream interface. Cleaning up the same interface twice (or more) here may raise
@@ -1300,6 +1364,13 @@
         pw.increaseIndent();
         dumpCounters(pw);
         pw.decreaseIndent();
+
+        pw.println();
+        pw.println("mSupportActiveSessionsMetrics: " + mSupportActiveSessionsMetrics);
+        pw.println("getLastMaxConnectionCount: "
+                + mBpfConntrackEventConsumer.getLastMaxConnectionCount());
+        pw.println("getCurrentConnectionCount: "
+                + mBpfConntrackEventConsumer.getCurrentConnectionCount());
     }
 
     private void dumpStats(@NonNull IndentingPrintWriter pw) {
@@ -1991,6 +2062,21 @@
     // while TCP status is established.
     @VisibleForTesting
     class BpfConntrackEventConsumer implements ConntrackEventConsumer {
+        /**
+         * Tracks the current number of tethering connections and the maximum
+         * observed since the last metrics collection. Used to provide insights
+         * into the distribution of active tethering sessions for metrics reporting.
+
+         * These variables are accessed on the handler thread, which includes:
+         *  1. ConntrackEvents signaling the addition or removal of an IPv4 rule.
+         *  2. ConntrackEvents indicating the removal of a tethering client,
+         *     triggering the removal of associated rules.
+         *  3. Removal of the last IpServer, which resets counters to handle
+         *     potential synchronization issues.
+         */
+        private int mLastMaxConnectionCount = 0;
+        private int mCurrentConnectionCount = 0;
+
         // The upstream4 and downstream4 rules are built as the following tables. Only raw ip
         // upstream interface is supported. Note that the field "lastUsed" is only updated by
         // BPF program which records the last used time for a given rule.
@@ -2124,6 +2210,10 @@
                     return;
                 }
 
+                if (mSupportActiveSessionsMetrics) {
+                    decreaseCurrentConnectionCount(1);
+                }
+
                 maybeClearLimit(upstreamIndex);
                 return;
             }
@@ -2136,8 +2226,50 @@
 
             maybeAddDevMap(upstreamIndex, tetherClient.downstreamIfindex);
             maybeSetLimit(upstreamIndex);
-            mBpfCoordinatorShim.tetherOffloadRuleAdd(UPSTREAM, upstream4Key, upstream4Value);
-            mBpfCoordinatorShim.tetherOffloadRuleAdd(DOWNSTREAM, downstream4Key, downstream4Value);
+
+            final boolean addedUpstream = mBpfCoordinatorShim.tetherOffloadRuleAdd(
+                    UPSTREAM, upstream4Key, upstream4Value);
+            final boolean addedDownstream = mBpfCoordinatorShim.tetherOffloadRuleAdd(
+                    DOWNSTREAM, downstream4Key, downstream4Value);
+            if (addedUpstream != addedDownstream) {
+                Log.wtf(TAG, "The bidirectional rules should be added concurrently ("
+                        + "upstream: " + addedUpstream
+                        + ", downstream: " + addedDownstream + ")");
+                return;
+            }
+            if (mSupportActiveSessionsMetrics && addedUpstream && addedDownstream) {
+                mCurrentConnectionCount++;
+                mLastMaxConnectionCount = Math.max(mCurrentConnectionCount,
+                        mLastMaxConnectionCount);
+            }
+        }
+
+        public int getLastMaxConnectionAndResetToCurrent() {
+            final int ret = mLastMaxConnectionCount;
+            mLastMaxConnectionCount = mCurrentConnectionCount;
+            return ret;
+        }
+
+        /** For dumping current state only. */
+        public int getLastMaxConnectionCount() {
+            return mLastMaxConnectionCount;
+        }
+
+        public int getCurrentConnectionCount() {
+            return mCurrentConnectionCount;
+        }
+
+        public void decreaseCurrentConnectionCount(int count) {
+            mCurrentConnectionCount -= count;
+            if (mCurrentConnectionCount < 0) {
+                Log.wtf(TAG, "Unexpected mCurrentConnectionCount: "
+                        + mCurrentConnectionCount);
+            }
+        }
+
+        public void clearConnectionCounters() {
+            mCurrentConnectionCount = 0;
+            mLastMaxConnectionCount = 0;
         }
     }
 
@@ -2477,6 +2609,11 @@
         });
     }
 
+    private void uploadConntrackMetricsSample() {
+        mDeps.sendTetheringActiveSessionsReported(
+                mBpfConntrackEventConsumer.getLastMaxConnectionAndResetToCurrent());
+    }
+
     private void schedulePollingStats() {
         if (mHandler.hasCallbacks(mScheduledPollingStats)) {
             mHandler.removeCallbacks(mScheduledPollingStats);
@@ -2494,6 +2631,15 @@
                 CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS);
     }
 
+    private void scheduleConntrackMetricsSampling() {
+        if (mHandler.hasCallbacks(mScheduledConntrackMetricsSampling)) {
+            mHandler.removeCallbacks(mScheduledConntrackMetricsSampling);
+        }
+
+        mHandler.postDelayed(mScheduledConntrackMetricsSampling,
+                CONNTRACK_METRICS_UPDATE_INTERVAL_MS);
+    }
+
     // Return IPv6 downstream forwarding rule map. This is used for testing only.
     // Note that this can be only called on handler thread.
     @NonNull
diff --git a/Tethering/src/com/android/networkstack/tethering/Tethering.java b/Tethering/src/com/android/networkstack/tethering/Tethering.java
index 13f4f2a..1938a08 100644
--- a/Tethering/src/com/android/networkstack/tethering/Tethering.java
+++ b/Tethering/src/com/android/networkstack/tethering/Tethering.java
@@ -374,6 +374,11 @@
                     }
 
                     @NonNull
+                    public Context getContext() {
+                        return mContext;
+                    }
+
+                    @NonNull
                     public INetd getNetd() {
                         return mNetd;
                     }
diff --git a/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java b/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
index 298940e..c9817c9 100644
--- a/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
+++ b/Tethering/src/com/android/networkstack/tethering/TetheringConfiguration.java
@@ -144,6 +144,12 @@
     /** A flag for using synchronous or asynchronous state machine. */
     public static boolean USE_SYNC_SM = false;
 
+    /**
+     * A feature flag to control whether the active sessions metrics should be enabled.
+     * Disabled by default.
+     */
+    public static final String TETHER_ACTIVE_SESSIONS_METRICS = "tether_active_sessions_metrics";
+
     public final String[] tetherableUsbRegexs;
     public final String[] tetherableWifiRegexs;
     public final String[] tetherableWigigRegexs;
diff --git a/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java b/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java
index e54a7e0..5d22977 100644
--- a/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java
+++ b/Tethering/tests/unit/src/com/android/networkstack/tethering/BpfCoordinatorTest.java
@@ -48,6 +48,7 @@
 import static com.android.net.module.util.netlink.StructNdMsg.NUD_FAILED;
 import static com.android.net.module.util.netlink.StructNdMsg.NUD_REACHABLE;
 import static com.android.net.module.util.netlink.StructNdMsg.NUD_STALE;
+import static com.android.networkstack.tethering.BpfCoordinator.CONNTRACK_METRICS_UPDATE_INTERVAL_MS;
 import static com.android.networkstack.tethering.BpfCoordinator.CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS;
 import static com.android.networkstack.tethering.BpfCoordinator.INVALID_MTU;
 import static com.android.networkstack.tethering.BpfCoordinator.NF_CONNTRACK_TCP_TIMEOUT_ESTABLISHED;
@@ -60,6 +61,7 @@
 import static com.android.networkstack.tethering.BpfUtils.DOWNSTREAM;
 import static com.android.networkstack.tethering.BpfUtils.UPSTREAM;
 import static com.android.networkstack.tethering.TetheringConfiguration.DEFAULT_TETHER_OFFLOAD_POLL_INTERVAL_MS;
+import static com.android.networkstack.tethering.TetheringConfiguration.TETHER_ACTIVE_SESSIONS_METRICS;
 import static com.android.testutils.MiscAsserts.assertSameElements;
 
 import static org.junit.Assert.assertArrayEquals;
@@ -87,6 +89,7 @@
 import static org.mockito.Mockito.when;
 
 import android.app.usage.NetworkStatsManager;
+import android.content.Context;
 import android.net.INetd;
 import android.net.InetAddresses;
 import android.net.IpPrefix;
@@ -140,6 +143,8 @@
 import com.android.testutils.DevSdkIgnoreRule.IgnoreUpTo;
 import com.android.testutils.TestBpfMap;
 import com.android.testutils.TestableNetworkStatsProviderCbBinder;
+import com.android.testutils.com.android.testutils.SetFeatureFlagsRule;
+import com.android.testutils.com.android.testutils.SetFeatureFlagsRule.FeatureFlag;
 
 import org.junit.Before;
 import org.junit.Rule;
@@ -171,6 +176,16 @@
     @Rule
     public final DevSdkIgnoreRule mIgnoreRule = new DevSdkIgnoreRule();
 
+    final HashMap<String, Boolean> mFeatureFlags = new HashMap<>();
+    // This will set feature flags from @FeatureFlag annotations
+    // into the map before setUp() runs.
+    @Rule
+    public final SetFeatureFlagsRule mSetFeatureFlagsRule =
+            new SetFeatureFlagsRule((name, enabled) -> {
+                mFeatureFlags.put(name, enabled);
+                return null;
+            }, (name) -> mFeatureFlags.getOrDefault(name, false));
+
     private static final boolean IPV4 = true;
     private static final boolean IPV6 = false;
 
@@ -406,6 +421,11 @@
                 return this;
             }
 
+            public Builder setPrivateAddress(Inet4Address privateAddr) {
+                mPrivateAddr = privateAddr;
+                return this;
+            }
+
             public Builder setRemotePort(int remotePort) {
                 mRemotePort = (short) remotePort;
                 return this;
@@ -429,6 +449,7 @@
 
     @Mock private NetworkStatsManager mStatsManager;
     @Mock private INetd mNetd;
+    @Mock private Context mMockContext;
     @Mock private IpServer mIpServer;
     @Mock private IpServer mIpServer2;
     @Mock private TetheringConfiguration mTetherConfig;
@@ -475,6 +496,11 @@
                     }
 
                     @NonNull
+                    public Context getContext() {
+                        return mMockContext;
+                    }
+
+                    @NonNull
                     public INetd getNetd() {
                         return mNetd;
                     }
@@ -546,6 +572,16 @@
                     public IBpfMap<S32, S32> getBpfErrorMap() {
                         return mBpfErrorMap;
                     }
+
+                    @Override
+                    public void sendTetheringActiveSessionsReported(int lastMaxSessionCount) {
+                        // No-op.
+                    }
+
+                    @Override
+                    public boolean isFeatureEnabled(Context context, String name) {
+                        return mFeatureFlags.getOrDefault(name, false);
+                    }
             });
 
     @Before public void setUp() {
@@ -1977,6 +2013,229 @@
         verify(mBpfDevMap, never()).updateEntry(any(), any());
     }
 
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS)
+    // BPF IPv4 forwarding only supports on S+.
+    @IgnoreUpTo(Build.VERSION_CODES.R)
+    @Test
+    public void testMaxConnectionCount_metricsEnabled() throws Exception {
+        doTestMaxConnectionCount(true);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS, enabled = false)
+    @Test
+    public void testMaxConnectionCount_metricsDisabled() throws Exception {
+        doTestMaxConnectionCount(false);
+    }
+
+    private void doTestMaxConnectionCount(final boolean supportActiveSessionsMetrics)
+            throws Exception {
+        final BpfCoordinator coordinator = makeBpfCoordinator();
+        initBpfCoordinatorForRule4(coordinator);
+        resetNetdAndBpfMaps();
+        assertEquals(0, mConsumer.getLastMaxConnectionAndResetToCurrent());
+
+        // Prepare add/delete rule events.
+        final ArrayList<ConntrackEvent> addRuleEvents = new ArrayList<>();
+        final ArrayList<ConntrackEvent> delRuleEvents = new ArrayList<>();
+        for (int i = 0; i < 5; i++) {
+            final ConntrackEvent addEvent = new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_NEW).setProto(IPPROTO_TCP).setRemotePort(i).build();
+            addRuleEvents.add(addEvent);
+            final ConntrackEvent delEvent = new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_DELETE).setProto(IPPROTO_TCP).setRemotePort(i).build();
+            delRuleEvents.add(delEvent);
+        }
+
+        // Add rules, verify counter increases.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(addRuleEvents.get(i));
+            assertConsumerCountersEquals(supportActiveSessionsMetrics ? i + 1 : 0);
+        }
+
+        // Add the same events again should not increase the counter because
+        // all events are already exist.
+        for (final ConntrackEvent event : addRuleEvents) {
+            mConsumer.accept(event);
+            assertConsumerCountersEquals(supportActiveSessionsMetrics ? 5 : 0);
+        }
+
+        // Verify removing non-existent items won't change the counters.
+        for (int i = 5; i < 8; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_DELETE).setProto(IPPROTO_TCP).setRemotePort(i).build());
+            assertConsumerCountersEquals(supportActiveSessionsMetrics ? 5 : 0);
+        }
+
+        // Verify remove the rules decrease the counter.
+        // Note the max counter returns the max, so it returns the count before deleting.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(delRuleEvents.get(i));
+            assertEquals(supportActiveSessionsMetrics ? 4 - i : 0,
+                    mConsumer.getCurrentConnectionCount());
+            assertEquals(supportActiveSessionsMetrics ? 5 - i : 0,
+                    mConsumer.getLastMaxConnectionCount());
+            assertEquals(supportActiveSessionsMetrics ? 5 - i : 0,
+                    mConsumer.getLastMaxConnectionAndResetToCurrent());
+        }
+
+        // Verify remove these rules again doesn't decrease the counter.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(delRuleEvents.get(i));
+            assertConsumerCountersEquals(0);
+        }
+    }
+
+    // Helper method to assert all counter values inside consumer.
+    private void assertConsumerCountersEquals(int expectedCount) {
+        assertEquals(expectedCount, mConsumer.getCurrentConnectionCount());
+        assertEquals(expectedCount, mConsumer.getLastMaxConnectionCount());
+        assertEquals(expectedCount, mConsumer.getLastMaxConnectionAndResetToCurrent());
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS)
+    // BPF IPv4 forwarding only supports on S+.
+    @IgnoreUpTo(Build.VERSION_CODES.R)
+    @Test
+    public void doTestMaxConnectionCount_removeClient_metricsEnabled() throws Exception {
+        doTestMaxConnectionCount_removeClient(true);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS, enabled = false)
+    @Test
+    public void doTestMaxConnectionCount_removeClient_metricsDisabled() throws Exception {
+        doTestMaxConnectionCount_removeClient(false);
+    }
+
+    private void doTestMaxConnectionCount_removeClient(final boolean supportActiveSessionsMetrics)
+            throws Exception {
+        final BpfCoordinator coordinator = makeBpfCoordinator();
+        initBpfCoordinatorForRule4(coordinator);
+        resetNetdAndBpfMaps();
+
+        // Add client information A and B on on the same downstream.
+        final ClientInfo clientA = new ClientInfo(DOWNSTREAM_IFINDEX, DOWNSTREAM_MAC,
+                PRIVATE_ADDR, MAC_A);
+        final ClientInfo clientB = new ClientInfo(DOWNSTREAM_IFINDEX, DOWNSTREAM_MAC,
+                PRIVATE_ADDR2, MAC_B);
+        coordinator.tetherOffloadClientAdd(mIpServer, clientA);
+        coordinator.tetherOffloadClientAdd(mIpServer, clientB);
+        assertClientInfoExists(mIpServer, clientA);
+        assertClientInfoExists(mIpServer, clientB);
+        assertEquals(0, mConsumer.getLastMaxConnectionAndResetToCurrent());
+
+        // Add some rules for both clients.
+        final int addr1RuleCount = 5;
+        final int addr2RuleCount = 3;
+
+        for (int i = 0; i < addr1RuleCount; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder()
+                    .setMsgType(IPCTNL_MSG_CT_NEW)
+                    .setProto(IPPROTO_TCP)
+                    .setRemotePort(i)
+                    .setPrivateAddress(PRIVATE_ADDR)
+                    .build());
+        }
+
+        for (int i = addr1RuleCount; i < addr1RuleCount + addr2RuleCount; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder()
+                    .setMsgType(IPCTNL_MSG_CT_NEW)
+                    .setProto(IPPROTO_TCP)
+                    .setRemotePort(i)
+                    .setPrivateAddress(PRIVATE_ADDR2)
+                    .build());
+        }
+
+        assertConsumerCountersEquals(
+                supportActiveSessionsMetrics ? addr1RuleCount + addr2RuleCount : 0);
+
+        // Remove 1 client. Since the 1st poll will return the LastMaxCounter and
+        // update it to the current, the max counter will be kept at 1st poll, while
+        // the current counter reflect the rule decreasing.
+        coordinator.tetherOffloadClientRemove(mIpServer, clientA);
+        assertEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0,
+                mConsumer.getCurrentConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr1RuleCount + addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr1RuleCount + addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionAndResetToCurrent());
+        // And all counters be updated at 2nd poll.
+        assertConsumerCountersEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0);
+
+        // Remove other client.
+        coordinator.tetherOffloadClientRemove(mIpServer, clientB);
+        assertEquals(0, mConsumer.getCurrentConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionCount());
+        assertEquals(supportActiveSessionsMetrics ? addr2RuleCount : 0,
+                mConsumer.getLastMaxConnectionAndResetToCurrent());
+        // All counters reach zero at 2nd poll.
+        assertConsumerCountersEquals(0);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS)
+    // BPF IPv4 forwarding only supports on S+.
+    @IgnoreUpTo(Build.VERSION_CODES.R)
+    @Test
+    public void testSendActiveSessionsReported_metricsEnabled() throws Exception {
+        doTestSendActiveSessionsReported(true);
+    }
+
+    @FeatureFlag(name = TETHER_ACTIVE_SESSIONS_METRICS, enabled = false)
+    @Test
+    public void testSendActiveSessionsReported_metricsDisabled() throws Exception {
+        doTestSendActiveSessionsReported(false);
+    }
+
+    private void doTestSendActiveSessionsReported(final boolean supportActiveSessionsMetrics)
+            throws Exception {
+        final BpfCoordinator coordinator = makeBpfCoordinator();
+        initBpfCoordinatorForRule4(coordinator);
+        resetNetdAndBpfMaps();
+        assertConsumerCountersEquals(0);
+
+        // Prepare the counter value.
+        for (int i = 0; i < 5; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_NEW).setProto(IPPROTO_TCP).setRemotePort(i).build());
+        }
+
+        // Then delete some 3 rules, 2 rules remaining.
+        // The max count is 5 while current rules count is 2.
+        for (int i = 0; i < 3; i++) {
+            mConsumer.accept(new TestConntrackEvent.Builder().setMsgType(
+                    IPCTNL_MSG_CT_DELETE).setProto(IPPROTO_TCP).setRemotePort(i).build());
+        }
+
+        // Verify the method is not invoked when timer is not expired.
+        waitForIdle();
+        verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+
+        // Verify metrics will be sent upon timer expiry.
+        mTestLooper.moveTimeForward(CONNTRACK_METRICS_UPDATE_INTERVAL_MS);
+        waitForIdle();
+        if (supportActiveSessionsMetrics) {
+            verify(mDeps).sendTetheringActiveSessionsReported(5);
+        } else {
+            verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+        }
+
+        // Verify next uploaded metrics will reflect the decreased rules count.
+        mTestLooper.moveTimeForward(CONNTRACK_METRICS_UPDATE_INTERVAL_MS);
+        waitForIdle();
+        if (supportActiveSessionsMetrics) {
+            verify(mDeps).sendTetheringActiveSessionsReported(2);
+        } else {
+            verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+        }
+
+        // Verify no metrics uploaded if polling stopped.
+        clearInvocations(mDeps);
+        coordinator.removeIpServer(mIpServer);
+        mTestLooper.moveTimeForward(CONNTRACK_TIMEOUT_UPDATE_INTERVAL_MS);
+        waitForIdle();
+        verify(mDeps, never()).sendTetheringActiveSessionsReported(anyInt());
+    }
+
     private void setElapsedRealtimeNanos(long nanoSec) {
         mElapsedRealtimeNanos = nanoSec;
     }
diff --git a/framework/Android.bp b/framework/Android.bp
index b30979a..0334e11 100644
--- a/framework/Android.bp
+++ b/framework/Android.bp
@@ -203,6 +203,7 @@
         "//packages/modules/Connectivity/tests:__subpackages__",
         "//packages/modules/Connectivity/thread/tests:__subpackages__",
         "//packages/modules/IPsec/tests/iketests",
+        "//packages/modules/NetworkStack",
         "//packages/modules/NetworkStack/tests:__subpackages__",
         "//packages/modules/Wifi/service/tests/wifitests",
     ],
diff --git a/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyDownloader.java b/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyDownloader.java
index f35b163..b2ef345 100644
--- a/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyDownloader.java
+++ b/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyDownloader.java
@@ -15,39 +15,68 @@
  */
 package com.android.server.net.ct;
 
+import android.annotation.RequiresApi;
 import android.app.DownloadManager;
 import android.content.BroadcastReceiver;
+import android.content.ContentResolver;
 import android.content.Context;
 import android.content.Intent;
 import android.content.IntentFilter;
 import android.net.Uri;
+import android.os.Build;
 import android.util.Log;
 
 import androidx.annotation.VisibleForTesting;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyFactory;
+import java.security.Signature;
+import java.security.spec.X509EncodedKeySpec;
+import java.util.Base64;
 
 /** Helper class to download certificate transparency log files. */
+@RequiresApi(Build.VERSION_CODES.VANILLA_ICE_CREAM)
 class CertificateTransparencyDownloader extends BroadcastReceiver {
 
     private static final String TAG = "CertificateTransparencyDownloader";
 
+    // TODO: move key to a DeviceConfig flag.
+    private static final byte[] PUBLIC_KEY_BYTES =
+            Base64.getDecoder()
+                    .decode(
+                            "MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAsu0BHGnQ++W2CTdyZyxv"
+                                + "HHRALOZPlnu/VMVgo2m+JZ8MNbAOH2cgXb8mvOj8flsX/qPMuKIaauO+PwROMjiq"
+                                + "fUpcFm80Kl7i97ZQyBDYKm3MkEYYpGN+skAR2OebX9G2DfDqFY8+jUpOOWtBNr3L"
+                                + "rmVcwx+FcFdMjGDlrZ5JRmoJ/SeGKiORkbbu9eY1Wd0uVhz/xI5bQb0OgII7hEj+"
+                                + "i/IPbJqOHgB8xQ5zWAJJ0DmG+FM6o7gk403v6W3S8qRYiR84c50KppGwe4YqSMkF"
+                                + "bLDleGQWLoaDSpEWtESisb4JiLaY4H+Kk0EyAhPSb+49JfUozYl+lf7iFN3qRq/S"
+                                + "IXXTh6z0S7Qa8EYDhKGCrpI03/+qprwy+my6fpWHi6aUIk4holUCmWvFxZDfixox"
+                                + "K0RlqbFDl2JXMBquwlQpm8u5wrsic1ksIv9z8x9zh4PJqNpCah0ciemI3YGRQqSe"
+                                + "/mRRXBiSn9YQBUPcaeqCYan+snGADFwHuXCd9xIAdFBolw9R9HTedHGUfVXPJDiF"
+                                + "4VusfX6BRR/qaadB+bqEArF/TzuDUr6FvOR4o8lUUxgLuZ/7HO+bHnaPFKYHHSm+"
+                                + "+z1lVDhhYuSZ8ax3T0C3FZpb7HMjZtpEorSV5ElKJEJwrhrBCMOD8L01EoSPrGlS"
+                                + "1w22i9uGHMn/uGQKo28u7AsCAwEAAQ==");
+
     private final Context mContext;
     private final DataStore mDataStore;
     private final DownloadHelper mDownloadHelper;
     private final CertificateTransparencyInstaller mInstaller;
+    private final byte[] mPublicKey;
 
     @VisibleForTesting
     CertificateTransparencyDownloader(
             Context context,
             DataStore dataStore,
             DownloadHelper downloadHelper,
-            CertificateTransparencyInstaller installer) {
+            CertificateTransparencyInstaller installer,
+            byte[] publicKey) {
         mContext = context;
         mDataStore = dataStore;
         mDownloadHelper = downloadHelper;
         mInstaller = installer;
+        mPublicKey = publicKey;
     }
 
     CertificateTransparencyDownloader(Context context, DataStore dataStore) {
@@ -55,13 +84,14 @@
                 context,
                 dataStore,
                 new DownloadHelper(context),
-                new CertificateTransparencyInstaller());
+                new CertificateTransparencyInstaller(),
+                PUBLIC_KEY_BYTES);
     }
 
     void registerReceiver() {
         IntentFilter intentFilter = new IntentFilter();
         intentFilter.addAction(DownloadManager.ACTION_DOWNLOAD_COMPLETE);
-        mContext.registerReceiver(this, intentFilter);
+        mContext.registerReceiver(this, intentFilter, Context.RECEIVER_EXPORTED);
 
         if (Config.DEBUG) {
             Log.d(TAG, "CertificateTransparencyDownloader initialized successfully");
@@ -139,12 +169,22 @@
             return;
         }
 
-        // TODO: 1. verify file signature, 2. validate file content.
+        boolean success = false;
+        try {
+            success = verify(contentUri, metadataUri);
+        } catch (IOException | GeneralSecurityException e) {
+            Log.e(TAG, "Could not verify new log list", e);
+        }
+        if (!success) {
+            Log.w(TAG, "Log list did not pass verification");
+            return;
+        }
+
+        // TODO: validate file content.
 
         String version = mDataStore.getProperty(Config.VERSION_PENDING);
         String contentUrl = mDataStore.getProperty(Config.CONTENT_URL_PENDING);
         String metadataUrl = mDataStore.getProperty(Config.METADATA_URL_PENDING);
-        boolean success = false;
         try (InputStream inputStream = mContext.getContentResolver().openInputStream(contentUri)) {
             success = mInstaller.install(inputStream, version);
         } catch (IOException e) {
@@ -161,6 +201,19 @@
         }
     }
 
+    private boolean verify(Uri file, Uri signature) throws IOException, GeneralSecurityException {
+        Signature verifier = Signature.getInstance("SHA256withRSA");
+        KeyFactory keyFactory = KeyFactory.getInstance("RSA");
+        verifier.initVerify(keyFactory.generatePublic(new X509EncodedKeySpec(mPublicKey)));
+        ContentResolver contentResolver = mContext.getContentResolver();
+
+        try (InputStream fileStream = contentResolver.openInputStream(file);
+                InputStream signatureStream = contentResolver.openInputStream(signature)) {
+            verifier.update(fileStream.readAllBytes());
+            return verifier.verify(signatureStream.readAllBytes());
+        }
+    }
+
     private long download(String url) {
         try {
             return mDownloadHelper.startDownload(url);
diff --git a/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyFlagsListener.java b/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyFlagsListener.java
index fdac434..f196abb 100644
--- a/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyFlagsListener.java
+++ b/networksecurity/service/src/com/android/server/net/ct/CertificateTransparencyFlagsListener.java
@@ -17,17 +17,18 @@
 
 import static android.provider.DeviceConfig.NAMESPACE_TETHERING;
 
+import android.annotation.RequiresApi;
 import android.content.Context;
+import android.os.Build;
 import android.provider.DeviceConfig;
 import android.provider.DeviceConfig.Properties;
 import android.text.TextUtils;
 import android.util.Log;
 
-import com.android.modules.utils.build.SdkLevel;
-
 import java.util.concurrent.Executors;
 
 /** Listener class for the Certificate Transparency Phenotype flags. */
+@RequiresApi(Build.VERSION_CODES.VANILLA_ICE_CREAM)
 class CertificateTransparencyFlagsListener implements DeviceConfig.OnPropertiesChangedListener {
 
     private static final String TAG = "CertificateTransparencyFlagsListener";
@@ -54,7 +55,7 @@
 
     @Override
     public void onPropertiesChanged(Properties properties) {
-        if (!SdkLevel.isAtLeastV() || !NAMESPACE_TETHERING.equals(properties.getNamespace())) {
+        if (!NAMESPACE_TETHERING.equals(properties.getNamespace())) {
             return;
         }
 
@@ -85,6 +86,8 @@
             return;
         }
 
+        // TODO: handle the case where there is already a pending download.
+
         mDataStore.setProperty(Config.VERSION_PENDING, newVersion);
         mDataStore.setProperty(Config.CONTENT_URL_PENDING, newContentUrl);
         mDataStore.setProperty(Config.METADATA_URL_PENDING, newMetadataUrl);
diff --git a/networksecurity/tests/unit/src/com/android/server/net/ct/CertificateTransparencyDownloaderTest.java b/networksecurity/tests/unit/src/com/android/server/net/ct/CertificateTransparencyDownloaderTest.java
index 5131a71..a056c35 100644
--- a/networksecurity/tests/unit/src/com/android/server/net/ct/CertificateTransparencyDownloaderTest.java
+++ b/networksecurity/tests/unit/src/com/android/server/net/ct/CertificateTransparencyDownloaderTest.java
@@ -40,7 +40,17 @@
 import org.mockito.MockitoAnnotations;
 
 import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.GeneralSecurityException;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
+import java.security.Signature;
 
 /** Tests for the {@link CertificateTransparencyDownloader}. */
 @RunWith(JUnit4.class)
@@ -49,15 +59,20 @@
     @Mock private DownloadHelper mDownloadHelper;
     @Mock private CertificateTransparencyInstaller mCertificateTransparencyInstaller;
 
+    private PrivateKey mPrivateKey;
     private Context mContext;
     private File mTempFile;
     private DataStore mDataStore;
     private CertificateTransparencyDownloader mCertificateTransparencyDownloader;
 
     @Before
-    public void setUp() throws IOException {
+    public void setUp() throws IOException, NoSuchAlgorithmException {
         MockitoAnnotations.initMocks(this);
 
+        KeyPairGenerator instance = KeyPairGenerator.getInstance("RSA");
+        KeyPair keyPair = instance.generateKeyPair();
+        mPrivateKey = keyPair.getPrivate();
+
         mContext = InstrumentationRegistry.getInstrumentation().getContext();
         mTempFile = File.createTempFile("datastore-test", ".properties");
         mDataStore = new DataStore(mTempFile);
@@ -65,7 +80,11 @@
 
         mCertificateTransparencyDownloader =
                 new CertificateTransparencyDownloader(
-                        mContext, mDataStore, mDownloadHelper, mCertificateTransparencyInstaller);
+                        mContext,
+                        mDataStore,
+                        mDownloadHelper,
+                        mCertificateTransparencyInstaller,
+                        keyPair.getPublic().getEncoded());
     }
 
     @After
@@ -128,23 +147,16 @@
     }
 
     @Test
-    public void testDownloader_handleContentCompleteInstallSuccessful() throws IOException {
+    public void testDownloader_handleContentCompleteInstallSuccessful() throws Exception {
         String version = "666";
-        mDataStore.setProperty(Config.VERSION_PENDING, version);
-
-        long metadataId = 123;
-        mDataStore.setPropertyLong(Config.METADATA_URL_KEY, metadataId);
-        Uri metadataUri = Uri.fromFile(File.createTempFile("log_list-metadata", "txt"));
-        mDataStore.setProperty(Config.METADATA_URL_PENDING, metadataUri.toString());
-        when(mDownloadHelper.getUri(metadataId)).thenReturn(metadataUri);
-
         long contentId = 666;
-        mDataStore.setPropertyLong(Config.CONTENT_URL_KEY, contentId);
-        when(mDownloadHelper.isSuccessful(contentId)).thenReturn(true);
-        Uri contentUri = Uri.fromFile(File.createTempFile("log_list", "json"));
-        mDataStore.setProperty(Config.CONTENT_URL_PENDING, contentUri.toString());
-        when(mDownloadHelper.getUri(contentId)).thenReturn(contentUri);
+        File logListFile = File.createTempFile("log_list", "json");
+        Uri contentUri = Uri.fromFile(logListFile);
+        long metadataId = 123;
+        File metadataFile = sign(logListFile);
+        Uri metadataUri = Uri.fromFile(metadataFile);
 
+        setUpDownloadComplete(version, metadataId, metadataUri, contentId, contentUri);
         when(mCertificateTransparencyInstaller.install(any(), eq(version))).thenReturn(true);
 
         assertThat(mDataStore.getProperty(Config.VERSION)).isNull();
@@ -161,23 +173,16 @@
     }
 
     @Test
-    public void testDownloader_handleContentCompleteInstallFails() throws IOException {
+    public void testDownloader_handleContentCompleteInstallFails() throws Exception {
         String version = "666";
-        mDataStore.setProperty(Config.VERSION_PENDING, version);
-
-        long metadataId = 123;
-        mDataStore.setPropertyLong(Config.METADATA_URL_KEY, metadataId);
-        Uri metadataUri = Uri.fromFile(File.createTempFile("log_list-metadata", "txt"));
-        mDataStore.setProperty(Config.METADATA_URL_PENDING, metadataUri.toString());
-        when(mDownloadHelper.getUri(metadataId)).thenReturn(metadataUri);
-
         long contentId = 666;
-        mDataStore.setPropertyLong(Config.CONTENT_URL_KEY, contentId);
-        when(mDownloadHelper.isSuccessful(contentId)).thenReturn(true);
-        Uri contentUri = Uri.fromFile(File.createTempFile("log_list", "json"));
-        mDataStore.setProperty(Config.CONTENT_URL_PENDING, contentUri.toString());
-        when(mDownloadHelper.getUri(contentId)).thenReturn(contentUri);
+        File logListFile = File.createTempFile("log_list", "json");
+        Uri contentUri = Uri.fromFile(logListFile);
+        long metadataId = 123;
+        File metadataFile = sign(logListFile);
+        Uri metadataUri = Uri.fromFile(metadataFile);
 
+        setUpDownloadComplete(version, metadataId, metadataUri, contentId, contentUri);
         when(mCertificateTransparencyInstaller.install(any(), eq(version))).thenReturn(false);
 
         mCertificateTransparencyDownloader.onReceive(
@@ -188,8 +193,56 @@
         assertThat(mDataStore.getProperty(Config.METADATA_URL)).isNull();
     }
 
+    @Test
+    public void testDownloader_handleContentCompleteVerificationFails() throws IOException {
+        String version = "666";
+        long contentId = 666;
+        Uri contentUri = Uri.fromFile(File.createTempFile("log_list", "json"));
+        long metadataId = 123;
+        Uri metadataUri = Uri.fromFile(File.createTempFile("log_list-wrong_metadata", "sig"));
+
+        setUpDownloadComplete(version, metadataId, metadataUri, contentId, contentUri);
+
+        mCertificateTransparencyDownloader.onReceive(
+                mContext, makeDownloadCompleteIntent(contentId));
+
+        verify(mCertificateTransparencyInstaller, never()).install(any(), eq(version));
+        assertThat(mDataStore.getProperty(Config.VERSION)).isNull();
+        assertThat(mDataStore.getProperty(Config.CONTENT_URL)).isNull();
+        assertThat(mDataStore.getProperty(Config.METADATA_URL)).isNull();
+    }
+
     private Intent makeDownloadCompleteIntent(long downloadId) {
         return new Intent(DownloadManager.ACTION_DOWNLOAD_COMPLETE)
                 .putExtra(DownloadManager.EXTRA_DOWNLOAD_ID, downloadId);
     }
+
+    private void setUpDownloadComplete(
+            String version, long metadataId, Uri metadataUri, long contentId, Uri contentUri)
+            throws IOException {
+        mDataStore.setProperty(Config.VERSION_PENDING, version);
+
+        mDataStore.setPropertyLong(Config.METADATA_URL_KEY, metadataId);
+        mDataStore.setProperty(Config.METADATA_URL_PENDING, metadataUri.toString());
+        when(mDownloadHelper.getUri(metadataId)).thenReturn(metadataUri);
+
+        mDataStore.setPropertyLong(Config.CONTENT_URL_KEY, contentId);
+        mDataStore.setProperty(Config.CONTENT_URL_PENDING, contentUri.toString());
+        when(mDownloadHelper.isSuccessful(contentId)).thenReturn(true);
+        when(mDownloadHelper.getUri(contentId)).thenReturn(contentUri);
+    }
+
+    private File sign(File file) throws IOException, GeneralSecurityException {
+        File signatureFile = File.createTempFile("log_list-metadata", "sig");
+        Signature signer = Signature.getInstance("SHA256withRSA");
+        signer.initSign(mPrivateKey);
+
+        try (InputStream fileStream = new FileInputStream(file);
+                OutputStream outputStream = new FileOutputStream(signatureFile)) {
+            signer.update(fileStream.readAllBytes());
+            outputStream.write(signer.sign());
+        }
+
+        return signatureFile;
+    }
 }
diff --git a/staticlibs/testutils/host/python/apf_test_base.py b/staticlibs/testutils/host/python/apf_test_base.py
index 7203265..9a30978 100644
--- a/staticlibs/testutils/host/python/apf_test_base.py
+++ b/staticlibs/testutils/host/python/apf_test_base.py
@@ -23,6 +23,11 @@
     super().setup_class()
 
     # Check test preconditions.
+    asserts.abort_class_if(
+        not self.client.isAtLeastV(),
+        "Do not enforce the test until V+ since chipset potential bugs are"
+        " expected to be fixed on V+ releases.",
+    )
     tether_utils.assume_hotspot_test_preconditions(
         self.serverDevice, self.clientDevice, UpstreamType.NONE
     )
@@ -34,13 +39,12 @@
     )
 
     # Fetch device properties and storing them locally for later use.
-    client = self.clientDevice.connectivity_multi_devices_snippet
     self.server_iface_name, client_network = (
         tether_utils.setup_hotspot_and_client_for_upstream_type(
             self.serverDevice, self.clientDevice, UpstreamType.NONE
         )
     )
-    self.client_iface_name = client.getInterfaceNameFromNetworkHandle(
+    self.client_iface_name = self.client.getInterfaceNameFromNetworkHandle(
         client_network
     )
     self.server_mac_address = apf_utils.get_hardware_address(
diff --git a/staticlibs/testutils/host/python/apf_utils.py b/staticlibs/testutils/host/python/apf_utils.py
index a3ec6e9..c3330d2 100644
--- a/staticlibs/testutils/host/python/apf_utils.py
+++ b/staticlibs/testutils/host/python/apf_utils.py
@@ -236,7 +236,7 @@
     ad: android_device.AndroidDevice, iface_name: str, expected_version: int
 ) -> None:
   caps = get_apf_capabilities(ad, iface_name)
-  asserts.skip_if(
+  asserts.abort_class_if(
       caps.apf_version_supported < expected_version,
       f"Supported apf version {caps.apf_version_supported} < expected version"
       f" {expected_version}",
diff --git a/staticlibs/testutils/host/python/multi_devices_test_base.py b/staticlibs/testutils/host/python/multi_devices_test_base.py
index f8a92f3..677329a 100644
--- a/staticlibs/testutils/host/python/multi_devices_test_base.py
+++ b/staticlibs/testutils/host/python/multi_devices_test_base.py
@@ -52,3 +52,4 @@
         max_workers=2,
         raise_on_exception=True,
     )
+    self.client = self.clientDevice.connectivity_multi_devices_snippet
diff --git a/tests/cts/multidevices/Android.bp b/tests/cts/multidevices/Android.bp
index 5f062f1..40aa1e4 100644
--- a/tests/cts/multidevices/Android.bp
+++ b/tests/cts/multidevices/Android.bp
@@ -26,6 +26,7 @@
         "run_tests.py",
     ],
     libs: [
+        "absl-py",
         "mobly",
         "net-tests-utils-host-python-common",
     ],
diff --git a/tests/cts/multidevices/apfv4_test.py b/tests/cts/multidevices/apfv4_test.py
index 4633d37..7795be5 100644
--- a/tests/cts/multidevices/apfv4_test.py
+++ b/tests/cts/multidevices/apfv4_test.py
@@ -12,23 +12,52 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from net_tests_utils.host.python import apf_test_base
+from absl.testing import parameterized
+from mobly import asserts
+from net_tests_utils.host.python import apf_test_base, apf_utils
 
 # Constants.
 COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED = "DROPPED_ETHERTYPE_NOT_ALLOWED"
 ETHER_BROADCAST_ADDR = "FFFFFFFFFFFF"
-ETH_P_ETHERCAT = "88A4"
 
 
-class ApfV4Test(apf_test_base.ApfTestBase):
+class ApfV4Test(apf_test_base.ApfTestBase, parameterized.TestCase):
+  def setup_class(self):
+    super().setup_class()
+    # Check apf version preconditions.
+    caps = apf_utils.get_apf_capabilities(
+        self.clientDevice, self.client_iface_name
+    )
+    if self.client.getVsrApiLevel() >= 34:
+      # Enforce APFv4 support for Android 14+ VSR.
+      asserts.assert_true(
+          caps.apf_version_supported >= 4,
+          "APFv4 became mandatory in Android 14 VSR.",
+      )
+    else:
+      # Skip tests for APF version < 4 before Android 14 VSR.
+      apf_utils.assume_apf_version_support_at_least(
+          self.clientDevice, self.client_iface_name, 4
+      )
 
-  def test_apf_drop_ethercat(self):
+  # APF L2 packet filtering on V+ Android allows only specific
+  # types: IPv4, ARP, IPv6, EAPOL, WAPI.
+  # Tests can use any disallowed packet type. Currently,
+  # several ethertypes from the legacy ApfFilter denylist are used.
+  @parameterized.parameters(
+      "88a2",  # ATA over Ethernet
+      "88a4",  # EtherCAT
+      "88b8",  # GOOSE (Generic Object Oriented Substation event)
+      "88cd",  # SERCOS III
+      "88e3",  # Media Redundancy Protocol (IEC62439-2)
+  )  # Declare inputs for state_str and expected_result.
+  def test_apf_drop_ethertype_not_allowed(self, blocked_ether_type):
     # Ethernet header (14 bytes).
     packet = ETHER_BROADCAST_ADDR  # Destination MAC (broadcast)
     packet += self.server_mac_address.replace(":", "")  # Source MAC
-    packet += ETH_P_ETHERCAT  # EtherType (EtherCAT)
+    packet += blocked_ether_type
 
-    # EtherCAT header (2 bytes) + 44 bytes of zero padding.
+    # Pad with zeroes to minimum ethernet frame length.
     packet += "00" * 46
     self.send_packet_and_expect_counter_increased(
         packet, COUNTER_DROPPED_ETHERTYPE_NOT_ALLOWED
diff --git a/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt b/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt
index 7368669..49688cc 100644
--- a/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt
+++ b/tests/cts/multidevices/snippet/ConnectivityMultiDevicesSnippet.kt
@@ -36,6 +36,7 @@
 import android.net.wifi.WifiNetworkSpecifier
 import android.net.wifi.WifiSsid
 import androidx.test.platform.app.InstrumentationRegistry
+import com.android.compatibility.common.util.PropertyUtil
 import com.android.modules.utils.build.SdkLevel
 import com.android.testutils.AutoReleaseNetworkCallbackRule
 import com.android.testutils.ConnectUtil
@@ -75,6 +76,12 @@
     @Rpc(description = "Check whether the device SDK is as least T")
     fun isAtLeastT() = SdkLevel.isAtLeastT()
 
+    @Rpc(description = "Return whether the Sdk level is at least V.")
+    fun isAtLeastV() = SdkLevel.isAtLeastV()
+
+    @Rpc(description = "Return the API level that the VSR requirement must be fulfilled.")
+    fun getVsrApiLevel() = PropertyUtil.getVsrApiLevel()
+
     @Rpc(description = "Request cellular connection and ensure it is the default network.")
     fun requestCellularAndEnsureDefault() {
         ctsNetUtils.disableWifi()