1*948ad32aSBojan Novković# 2*948ad32aSBojan Novković# Copyright (c) 2026 Stormshield 3*948ad32aSBojan Novković# 4*948ad32aSBojan Novković# SPDX-License-Identifier: BSD-2-Clause 5*948ad32aSBojan Novković# 6*948ad32aSBojan Novković 7*948ad32aSBojan Novkovićimport pytest 8*948ad32aSBojan Novkovićimport errno 9*948ad32aSBojan Novkovićimport os 10*948ad32aSBojan Novkovićimport socket 11*948ad32aSBojan Novkovićimport struct 12*948ad32aSBojan Novkovićimport sys 13*948ad32aSBojan Novkovićimport logging 14*948ad32aSBojan Novković 15*948ad32aSBojan Novkovićfrom atf_python.sys.net.vnet import VnetTestTemplate 16*948ad32aSBojan Novkovićfrom atf_python.sys.net.tools import ToolsHelper 17*948ad32aSBojan Novković 18*948ad32aSBojan Novkovićdef _common(tunable, addr, domain, type, proto): 19*948ad32aSBojan Novković """ 20*948ad32aSBojan Novković Test what happens when we try to bind a socket to an 21*948ad32aSBojan Novković address that is not present in the current FIB. 22*948ad32aSBojan Novković """ 23*948ad32aSBojan Novković sysctl_output = ToolsHelper.get_output(f"sysctl {tunable}") 24*948ad32aSBojan Novković tunable_val = int(sysctl_output.split(":")[1]) 25*948ad32aSBojan Novković if tunable_val != 0: 26*948ad32aSBojan Novković pytest.skip(f"{tunable} must be set to 0") 27*948ad32aSBojan Novković 28*948ad32aSBojan Novković port = 12345 if type != socket.SOCK_RAW else 0 29*948ad32aSBojan Novković s = socket.socket(domain, type, proto) 30*948ad32aSBojan Novković s.setsockopt(socket.SOL_SOCKET, socket.SO_SETFIB, 0) 31*948ad32aSBojan Novković 32*948ad32aSBojan Novković failed = False 33*948ad32aSBojan Novković try: 34*948ad32aSBojan Novković s.bind((addr, port)) 35*948ad32aSBojan Novković except OSError as e: 36*948ad32aSBojan Novković assert e.errno == errno.EADDRNOTAVAIL 37*948ad32aSBojan Novković failed = True 38*948ad32aSBojan Novković assert failed 39*948ad32aSBojan Novković 40*948ad32aSBojan Novković s.setsockopt(socket.SOL_SOCKET, socket.SO_SETFIB, 1) 41*948ad32aSBojan Novković 42*948ad32aSBojan Novković failed = False 43*948ad32aSBojan Novković try: 44*948ad32aSBojan Novković s.bind((addr, port)) 45*948ad32aSBojan Novković except OSError: 46*948ad32aSBojan Novković failed = True 47*948ad32aSBojan Novković assert not failed 48*948ad32aSBojan Novković s.close() 49*948ad32aSBojan Novković 50*948ad32aSBojan Novkovićclass TestFibBind(VnetTestTemplate): 51*948ad32aSBojan Novković REQUIRED_MODULES = [] 52*948ad32aSBojan Novković TOPOLOGY = { 53*948ad32aSBojan Novković "vnet1": { "ifaces": [ "if1", "if2" ] }, 54*948ad32aSBojan Novković "if1": { 55*948ad32aSBojan Novković "prefixes4": [("192.168.1.1/24", "192.168.1.2/24")], 56*948ad32aSBojan Novković "fib": (0, 0), 57*948ad32aSBojan Novković }, 58*948ad32aSBojan Novković "if2": { 59*948ad32aSBojan Novković "prefixes4": [("192.168.2.1/24", "192.168.2.2/24")], 60*948ad32aSBojan Novković "fib": (1, 0), 61*948ad32aSBojan Novković }, 62*948ad32aSBojan Novković } 63*948ad32aSBojan Novković 64*948ad32aSBojan Novković def setup_method(self, method): 65*948ad32aSBojan Novković super().setup_method(method) 66*948ad32aSBojan Novković 67*948ad32aSBojan Novković def test_TCP(self): 68*948ad32aSBojan Novković _common("net.inet.tcp.bind_all_fibs", "192.168.2.1", socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP) 69*948ad32aSBojan Novković 70*948ad32aSBojan Novković def test_UDP(self): 71*948ad32aSBojan Novković _common("net.inet.udp.bind_all_fibs", "192.168.2.1", socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) 72*948ad32aSBojan Novković 73*948ad32aSBojan Novković def test_RAW(self): 74*948ad32aSBojan Novković _common("net.inet.raw.bind_all_fibs", "192.168.2.1", socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP) 75*948ad32aSBojan Novković 76*948ad32aSBojan Novkovićclass TestFibBind6(VnetTestTemplate): 77*948ad32aSBojan Novković REQUIRED_MODULES = [] 78*948ad32aSBojan Novković TOPOLOGY = { 79*948ad32aSBojan Novković "vnet1": { "ifaces": [ "if1", "if2" ] }, 80*948ad32aSBojan Novković "if1": { 81*948ad32aSBojan Novković "prefixes6": [("2001:db8:0:1::1/64", "2001:db8:0:1::2/64")], 82*948ad32aSBojan Novković "fib": (0, 0), 83*948ad32aSBojan Novković }, 84*948ad32aSBojan Novković "if2": { 85*948ad32aSBojan Novković "prefixes6": [("2001:db8:0:2::1/64", "2001:db8:0:2::2/64")], 86*948ad32aSBojan Novković "fib": (1, 0), 87*948ad32aSBojan Novković }, 88*948ad32aSBojan Novković } 89*948ad32aSBojan Novković 90*948ad32aSBojan Novković def setup_method(self, method): 91*948ad32aSBojan Novković super().setup_method(method) 92*948ad32aSBojan Novković 93*948ad32aSBojan Novković def test_TCP(self): 94*948ad32aSBojan Novković _common("net.inet.tcp.bind_all_fibs", "2001:db8:0:2::1", socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP) 95*948ad32aSBojan Novković 96*948ad32aSBojan Novković def test_UDP(self): 97*948ad32aSBojan Novković _common("net.inet.udp.bind_all_fibs", "2001:db8:0:2::1", socket.AF_INET6, socket.SOCK_DGRAM, socket.IPPROTO_UDP) 98*948ad32aSBojan Novković 99*948ad32aSBojan Novković def test_RAW(self): 100*948ad32aSBojan Novković _common("net.inet.raw.bind_all_fibs", "2001:db8:0:2::1", socket.AF_INET6, socket.SOCK_RAW, socket.IPPROTO_IPV6) 101