diff --git a/src/plugins/acl/fa_node.c b/src/plugins/acl/fa_node.c
index b706fce..c71429e 100644
--- a/src/plugins/acl/fa_node.c
+++ b/src/plugins/acl/fa_node.c
@@ -570,10 +570,11 @@
 }
 
 static void
-acl_fa_conn_list_add_session (acl_main_t * am, u32 sess_id)
+acl_fa_conn_list_add_session (acl_main_t * am, u32 sess_id, u64 now)
 {
   fa_session_t *sess = am->fa_sessions_pool + sess_id;
   u8 list_id = fa_session_get_timeout_type(am, sess);
+  sess->link_enqueue_time = now;
   sess->link_list_id = list_id;
   sess->link_next_idx = ~0;
   sess->link_prev_idx = am->fa_conn_list_tail[list_id];
@@ -629,7 +630,7 @@
 {
   // fa_session_t *sess = am->fa_sessions_pool + sess_id;
   acl_fa_conn_list_delete_session(am, sess_id);
-  acl_fa_conn_list_add_session(am, sess_id);
+  acl_fa_conn_list_add_session(am, sess_id, now);
 }
 
 
@@ -720,7 +721,7 @@
 
   BV (clib_bihash_add_del) (&am->fa_sessions_by_sw_if_index[sw_if_index],
 			    &kv, 1);
-  acl_fa_conn_list_add_session(am, sess_id);
+  acl_fa_conn_list_add_session(am, sess_id, now);
 
   vec_validate (am->fa_session_adds_by_sw_if_index, sw_if_index);
   am->fa_session_adds_by_sw_if_index[sw_if_index]++;
@@ -1097,12 +1098,12 @@
 static vlib_node_registration_t acl_fa_session_cleaner_process_node;
 
 static int
-acl_fa_conn_has_timed_out (acl_main_t *am, u64 now, u32 session_index)
+acl_fa_conn_time_to_check (acl_main_t *am, u64 now, u32 session_index)
 {
   fa_session_t *sess = am->fa_sessions_pool + session_index;
-  u64 sess_timeout_time =
-              sess->last_active_time + fa_session_get_timeout (am, sess);
-  return (sess_timeout_time < now);
+  u64 timeout_time =
+              sess->link_enqueue_time + fa_session_get_timeout (am, sess);
+  return (timeout_time < now);
 }
 
 
@@ -1210,7 +1211,7 @@
         for(tt = 0; tt < ACL_N_TIMEOUTS; tt++) {
           while((vec_len(expired) < 2*am->fa_max_deleted_sessions_per_interval)
                 && (~0 != am->fa_conn_list_head[tt])
-                && (acl_fa_conn_has_timed_out(am, now,
+                && (acl_fa_conn_time_to_check(am, now,
                                               am->fa_conn_list_head[tt]))) {
             u32 sess_id = am->fa_conn_list_head[tt];
             vec_add1(expired, sess_id);
@@ -1237,7 +1238,7 @@
                 /* There was activity on the session, so the idle timeout
                    has not passed. Enqueue for another time period. */
 
-                acl_fa_conn_list_add_session(am, session_index);
+                acl_fa_conn_list_add_session(am, session_index, now);
 
 		/* FIXME: When/if moving to timer wheel,
                    pretend we did this in the past,
diff --git a/src/plugins/acl/fa_node.h b/src/plugins/acl/fa_node.h
index 8edd006..8618362 100644
--- a/src/plugins/acl/fa_node.h
+++ b/src/plugins/acl/fa_node.h
@@ -63,7 +63,8 @@
   u8 reserved1;           /* +1 bytes = 64 */
   u32 link_prev_idx;
   u32 link_next_idx;
-  u64 reserved2[7];
+  u64 link_enqueue_time;
+  u64 reserved2[6];
 } fa_session_t;
 
 
diff --git a/test/test_acl_plugin_conns.py b/test/test_acl_plugin_conns.py
new file mode 100644
index 0000000..be016d9
--- /dev/null
+++ b/test/test_acl_plugin_conns.py
@@ -0,0 +1,304 @@
+#!/usr/bin/env python
+""" ACL plugin extended stateful tests """
+
+import unittest
+from framework import VppTestCase, VppTestRunner, running_extended_tests
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
+from scapy.layers.inet import IP, UDP
+from scapy.packet import Packet
+from socket import inet_pton, AF_INET, AF_INET6
+from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest
+from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting
+from scapy.layers.inet6 import IPv6ExtHdrFragment
+from pprint import pprint
+from random import randint
+
+
+def to_acl_rule(self, is_permit, wildcard_sport=False):
+    p = self
+    rule_family = AF_INET6 if p.haslayer(IPv6) else AF_INET
+    rule_prefix_len = 128 if p.haslayer(IPv6) else 32
+    rule_l3_layer = IPv6 if p.haslayer(IPv6) else IP
+    rule_l4_sport = p.sport
+    rule_l4_dport = p.dport
+    if p.haslayer(IPv6):
+        rule_l4_proto = p[IPv6].nh
+    else:
+        rule_l4_proto = p[IP].proto
+
+    if wildcard_sport:
+        rule_l4_sport_first = 0
+        rule_l4_sport_last = 65535
+    else:
+        rule_l4_sport_first = rule_l4_sport
+        rule_l4_sport_last = rule_l4_sport
+
+    new_rule = {
+          'is_permit': is_permit,
+          'is_ipv6': p.haslayer(IPv6),
+          'src_ip_addr': inet_pton(rule_family,
+                                   p[rule_l3_layer].src),
+          'src_ip_prefix_len': rule_prefix_len,
+          'dst_ip_addr': inet_pton(rule_family,
+                                   p[rule_l3_layer].dst),
+          'dst_ip_prefix_len': rule_prefix_len,
+          'srcport_or_icmptype_first': rule_l4_sport_first,
+          'srcport_or_icmptype_last': rule_l4_sport_last,
+          'dstport_or_icmpcode_first': rule_l4_dport,
+          'dstport_or_icmpcode_last': rule_l4_dport,
+          'proto': rule_l4_proto,
+         }
+    return new_rule
+
+Packet.to_acl_rule = to_acl_rule
+
+
+class IterateWithSleep():
+    def __init__(self, testcase, n_iters, description, sleep_sec):
+        self.curr = 0
+        self.testcase = testcase
+        self.n_iters = n_iters
+        self.sleep_sec = sleep_sec
+        self.description = description
+
+    def __iter__(self):
+        for x in range(0, self.n_iters):
+            yield x
+            self.testcase.sleep(self.sleep_sec)
+
+
+class Conn():
+    def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
+        self.testcase = testcase
+        self.ifs = [None, None]
+        self.ifs[0] = if1
+        self.ifs[1] = if2
+        self.address_family = af
+        self.l4proto = l4proto
+        self.ports = [None, None]
+        self.ports[0] = port1
+        self.ports[1] = port2
+        self
+
+    def pkt(self, side):
+        is_ip6 = 1 if self.address_family == AF_INET6 else 0
+        s0 = side
+        s1 = 1-side
+        src_if = self.ifs[s0]
+        dst_if = self.ifs[s1]
+        layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
+                   IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)]
+        payload = "x"
+        p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
+             layer_3[is_ip6] /
+             self.l4proto(sport=self.ports[s0], dport=self.ports[s1]) /
+             Raw(payload))
+        return p
+
+    def apply_acls(self, reflect_side, acl_side):
+        pkts = []
+        pkts.append(self.pkt(0))
+        pkts.append(self.pkt(1))
+        pkt = pkts[reflect_side]
+
+        r = []
+        r.append(pkt.to_acl_rule(2, wildcard_sport=True))
+        r.append(self.wildcard_rule(0))
+        res = self.testcase.api_acl_add_replace(0xffffffff, r)
+        self.testcase.assert_equal(res.retval, 0, "error adding ACL")
+        reflect_acl_index = res.acl_index
+
+        r = []
+        r.append(self.wildcard_rule(0))
+        res = self.testcase.api_acl_add_replace(0xffffffff, r)
+        self.testcase.assert_equal(res.retval, 0, "error adding deny ACL")
+        deny_acl_index = res.acl_index
+
+        if reflect_side == acl_side:
+            self.testcase.api_acl_interface_set_acl_list(
+                   self.ifs[acl_side].sw_if_index, 2, 1,
+                   [reflect_acl_index,
+                    deny_acl_index])
+            self.testcase.api_acl_interface_set_acl_list(
+                   self.ifs[1-acl_side].sw_if_index, 0, 0, [])
+        else:
+            self.testcase.api_acl_interface_set_acl_list(
+                   self.ifs[acl_side].sw_if_index, 2, 1,
+                   [deny_acl_index,
+                    reflect_acl_index])
+            self.testcase.api_acl_interface_set_acl_list(
+                   self.ifs[1-acl_side].sw_if_index, 0, 0, [])
+
+    def wildcard_rule(self, is_permit):
+        any_addr = ["0.0.0.0", "::"]
+        rule_family = self.address_family
+        is_ip6 = 1 if rule_family == AF_INET6 else 0
+        new_rule = {
+              'is_permit': is_permit,
+              'is_ipv6': is_ip6,
+              'src_ip_addr': inet_pton(rule_family, any_addr[is_ip6]),
+              'src_ip_prefix_len': 0,
+              'dst_ip_addr': inet_pton(rule_family, any_addr[is_ip6]),
+              'dst_ip_prefix_len': 0,
+              'srcport_or_icmptype_first': 0,
+              'srcport_or_icmptype_last': 65535,
+              'dstport_or_icmpcode_first': 0,
+              'dstport_or_icmpcode_last': 65535,
+              'proto': 0,
+             }
+        return new_rule
+
+    def send(self, side):
+        self.ifs[side].add_stream(self.pkt(side))
+        self.ifs[1-side].enable_capture()
+        self.testcase.pg_start()
+
+    def recv(self, side):
+        p = self.ifs[side].wait_for_packet(1)
+        return p
+
+    def send_through(self, side):
+        self.send(side)
+        p = self.recv(1-side)
+        return p
+
+    def send_pingpong(self, side):
+        p1 = self.send_through(side)
+        p2 = self.send_through(1-side)
+        return [p1, p2]
+
+
+@unittest.skipUnless(running_extended_tests(), "part of extended tests")
+class ACLPluginConnTestCase(VppTestCase):
+    """ ACL plugin connection-oriented extended testcases """
+
+    @classmethod
+    def setUpClass(self):
+        super(ACLPluginConnTestCase, self).setUpClass()
+        # create pg0 and pg1
+        self.create_pg_interfaces(range(2))
+        for i in self.pg_interfaces:
+            i.admin_up()
+            i.config_ip4()
+            i.config_ip6()
+            i.resolve_arp()
+            i.resolve_ndp()
+
+    def api_acl_add_replace(self, acl_index, r, count=-1, tag="",
+                            expected_retval=0):
+        """Add/replace an ACL
+
+        :param int acl_index: ACL index to replace, 4294967295 to create new.
+        :param acl_rule r: ACL rules array.
+        :param str tag: symbolic tag (description) for this ACL.
+        :param int count: number of rules.
+        """
+        if (count < 0):
+            count = len(r)
+        return self.vapi.api(self.vapi.papi.acl_add_replace,
+                             {'acl_index': acl_index,
+                              'r': r,
+                              'count': count,
+                              'tag': tag
+                              }, expected_retval=expected_retval)
+
+    def api_acl_interface_set_acl_list(self, sw_if_index, count, n_input, acls,
+                                       expected_retval=0):
+        return self.vapi.api(self.vapi.papi.acl_interface_set_acl_list,
+                             {'sw_if_index': sw_if_index,
+                              'count': count,
+                              'n_input': n_input,
+                              'acls': acls
+                              }, expected_retval=expected_retval)
+
+    def api_acl_dump(self, acl_index, expected_retval=0):
+        return self.vapi.api(self.vapi.papi.acl_dump,
+                             {'acl_index': acl_index},
+                             expected_retval=expected_retval)
+
+    def run_basic_conn_test(self, af, acl_side):
+        """ Basic conn timeout test """
+        conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242)
+        conn1.apply_acls(0, acl_side)
+        conn1.send_through(0)
+        # the return packets should pass
+        conn1.send_through(1)
+        # send some packets on conn1, ensure it doesn't go away
+        for i in IterateWithSleep(self, 20, "Keep conn active", 0.3):
+            conn1.send_through(1)
+        # allow the conn to time out
+        for i in IterateWithSleep(self, 30, "Wait for timeout", 0.1):
+            pass
+        # now try to send a packet on the reflected side
+        try:
+            p2 = conn1.send_through(1).command()
+        except:
+            # If we asserted while waiting, it's good.
+            # the conn should have timed out.
+            p2 = None
+        self.assert_equal(p2, None, "packet on long-idle conn")
+
+    def run_active_conn_test(self, af, acl_side):
+        """ Idle connection behind active connection test """
+        base = 10000 + 1000*acl_side
+        conn1 = Conn(self, self.pg0, self.pg1, af, UDP, base + 1, 2323)
+        conn2 = Conn(self, self.pg0, self.pg1, af, UDP, base + 2, 2323)
+        conn3 = Conn(self, self.pg0, self.pg1, af, UDP, base + 3, 2323)
+        conn1.apply_acls(0, acl_side)
+        conn1.send(0)
+        conn1.recv(1)
+        # create and check that the conn2/3 work
+        self.sleep(0.1)
+        conn2.send_pingpong(0)
+        self.sleep(0.1)
+        conn3.send_pingpong(0)
+        # send some packets on conn1, keep conn2/3 idle
+        for i in IterateWithSleep(self, 20, "Keep conn active", 0.2):
+            conn1.send_through(1)
+        try:
+            p2 = conn2.send_through(1).command()
+        except:
+            # If we asserted while waiting, it's good.
+            # the conn should have timed out.
+            p2 = None
+        # We should have not received the packet on a long-idle
+        # connection, because it should have timed out
+        # If it didn't - it is a problem
+        self.assert_equal(p2, None, "packet on long-idle conn")
+
+    def test_0000_conn_prepare_test(self):
+        """ Prepare the settings """
+        self.vapi.ppcli("set acl-plugin session timeout udp idle 1")
+
+    def test_0001_basic_conn_test(self):
+        """ IPv4: Basic conn timeout test reflect on ingress """
+        self.run_basic_conn_test(AF_INET, 0)
+
+    def test_0002_basic_conn_test(self):
+        """ IPv4: Basic conn timeout test reflect on egress """
+        self.run_basic_conn_test(AF_INET, 1)
+
+    def test_0011_active_conn_test(self):
+        """ IPv4: Idle conn behind active conn, reflect on ingress """
+        self.run_active_conn_test(AF_INET, 0)
+
+    def test_0012_active_conn_test(self):
+        """ IPv4: Idle conn behind active conn, reflect on egress """
+        self.run_active_conn_test(AF_INET, 1)
+
+    def test_1001_basic_conn_test(self):
+        """ IPv6: Basic conn timeout test reflect on ingress """
+        self.run_basic_conn_test(AF_INET6, 0)
+
+    def test_1002_basic_conn_test(self):
+        """ IPv6: Basic conn timeout test reflect on egress """
+        self.run_basic_conn_test(AF_INET6, 1)
+
+    def test_1011_active_conn_test(self):
+        """ IPv6: Idle conn behind active conn, reflect on ingress """
+        self.run_active_conn_test(AF_INET6, 0)
+
+    def test_1012_active_conn_test(self):
+        """ IPv6: Idle conn behind active conn, reflect on egress """
+        self.run_active_conn_test(AF_INET6, 1)