diff --git a/test/local/tcp_tester.py b/test/local/tcp_tester.py
index 689fb346..ff654d3a 100644
--- a/test/local/tcp_tester.py
+++ b/test/local/tcp_tester.py
@@ -9,6 +9,10 @@ class _TCPTester:
 	TCP_NORMAL_REQUEST = "Hello"
 	TCP_NORMAL_RESPONSE = "Same to you"
 
+	TCP_SYN_NORMAL = 0
+	TCP_SYN_RETRANSMIT = 1
+	TCP_SYN_SCAN = 2
+
 	def __init__(self, client_tap, client_mac, client_ip, client_port,
 					   server_tap, server_mac, server_ip, server_port,
 					   client_pkt_check=None, server_pkt_check=None):
@@ -40,7 +44,7 @@ def get_server_packet(self):
 			self.server_pkt_check(pkt)
 		return pkt
 
-	def reply_tcp(self):
+	def reply_tcp(self, syn_style):
 		pkt = self.get_server_packet()
 
 		# Received ACK only, just end
@@ -62,6 +66,10 @@ def reply_tcp(self):
 					 TCP(dport=pkt[TCP].sport, sport=pkt[TCP].dport, seq=self.tcp_receiver_seq, flags=flags, ack=pkt[TCP].seq+1, options=[("NOP", None)]))
 		delayed_sendp(reply_pkt, self.server_tap)
 
+		if syn_style == _TCPTester.TCP_SYN_RETRANSMIT:
+			delayed_sendp(reply_pkt, self.server_tap)
+			return
+
 		if flags != "A":
 			self.tcp_receiver_seq += 1
 
@@ -84,6 +92,9 @@ def reply_tcp(self):
 				self.tcp_receiver_seq += len(_TCPTester.TCP_NORMAL_RESPONSE)
 				# and continue with ACK
 
+		if syn_style != _TCPTester.TCP_SYN_NORMAL:
+			return
+
 		# Await ACK
 		pkt = self.get_server_packet()
 		assert pkt[TCP].flags == "A", \
@@ -104,8 +115,8 @@ def get_client_packet(self):
 			client_pkt_check(pkt)
 		return pkt
 
-	def request_tcp(self, flags, payload=None):
-		server_thread = threading.Thread(target=self.reply_tcp)
+	def request_tcp(self, flags, payload=None, syn_style=TCP_SYN_NORMAL):
+		server_thread = threading.Thread(target=self.reply_tcp, args=(syn_style,))
 		server_thread.start()
 
 		tcp_pkt = (Ether(dst=self.server_mac, src=self.client_mac, type=0x0800) /
@@ -148,6 +159,9 @@ def request_tcp(self, flags, payload=None):
 					"Bad answer from server"
 			reply_seq += len(payload)
 
+		if syn_style != _TCPTester.TCP_SYN_NORMAL:
+			return
+
 		# send ACK
 		tcp_pkt = (Ether(dst=self.server_mac, src=self.client_mac, type=0x0800) /
 				   IP(dst=self.server_ip, src=self.client_ip) /
@@ -180,6 +194,17 @@ def leave_open(self):
 		self.reset()
 		self.request_tcp("S")
 
+	# Helper function to simulate a SYN port scan
+	def syn_scan(self):
+		self.reset()
+		self.request_tcp("S", syn_style=_TCPTester.TCP_SYN_SCAN)
+		self.request_tcp("S", syn_style=_TCPTester.TCP_SYN_SCAN)
+
+	# Helper function to simulate a SYNACK being retransmitted
+	def syn_retrans(self):
+		self.reset()
+		self.request_tcp("S", syn_style=_TCPTester.TCP_SYN_RETRANSMIT)
+
 
 class TCPTesterLocal(_TCPTester):
 	def __init__(self, client_vm, client_port, server_vm, server_port, client_pkt_check=None, server_pkt_check=None):
diff --git a/test/local/xtratest_flow_timeout.py b/test/local/xtratest_flow_timeout.py
index 91bef6d0..e7b0ad52 100644
--- a/test/local/xtratest_flow_timeout.py
+++ b/test/local/xtratest_flow_timeout.py
@@ -185,3 +185,34 @@ def test_external_lb_relay_algorithm(prepare_ipv4, grpc_client, fast_flow_timeou
 	for target in targets:
 		grpc_client.dellbtarget(lb_name, target)
 	grpc_client.dellb(lb_name)
+
+
+def test_syn_scan(request, prepare_ipv4, grpc_client, fast_flow_timeout):
+	if not fast_flow_timeout:
+		pytest.skip("Fast flow timeout needs to be enabled")
+
+	# Only allow one port for this test, so the next call would normally fail (NAT runs out of free ports)
+	nat_ul_ipv6 = grpc_client.addnat(VM1.name, nat_vip, nat_local_min_port, nat_local_min_port+1)
+
+	# This produces SYN-SYNACK-SYN-SYNACK
+	tester = TCPTesterPublic(VM1, 12344, nat_ul_ipv6, PF0, public_ip, 443)
+	tester.syn_scan()
+
+	age_out_flows()
+
+	# (the only) NAT port should once again be free now
+	tester.client_port = 54321
+	tester.request_rst()
+
+	age_out_flows()
+
+	# This produces SYN-SYNACK+SYNACK
+	tester.client_port = 12345
+	tester.syn_retrans()
+
+	age_out_flows()
+
+	tester.client_port = 54321
+	tester.request_rst()
+
+	grpc_client.delnat(VM1.name)