Code update py311

examples/README.txt

@@ -22,5 +22,5 @@ Before running any of the examples, please make sure to:
  - Run the 'sudo make tap' command to create the tap7 interface and assign it to the 'br0' bridge.
  - Run the 'make' command to create the proper virtual environment.
  - Run '. venv/bin/activate' command to start the stack virtual environment.
- - Execute any example, e.g., 'example/run_stack.py'.
- - Hit Ctrl-C to stop it.
+ - Execute any example, e.g., 'examples/run_stack.py'.
+ - Hit Ctrl-C to stop it.

examples/icmp_echo_client.py

@@ -43,7 +43,7 @@
 
 import click
 
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 from pytcp.lib import stack
 from pytcp.lib.ip4_address import Ip4Address
 from pytcp.lib.ip6_address import Ip6Address
@@ -145,7 +145,7 @@ def cli(*, interface: str, remote_ip_address: str) -> None:
     Run the ICMP Echo client.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     client = IcmpEchoClient(
         remote_ip_address=remote_ip_address,
     )

examples/run_stack.py

@@ -37,7 +37,7 @@
 
 import click
 
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 
 
 @click.command()
@@ -61,7 +61,7 @@ def cli(
     """
 
     stack = TcpIpStack(
-        interface=interface,
+        fd=initialize_tap(tap_name=interface),
         mac_address=mac_address,
         ip6_address=ip6_address,
         ip6_gateway=ip6_gateway,

examples/tcp_daytime_service.py

@@ -42,7 +42,7 @@
 import click
 
 from examples.lib.tcp_service import TcpService
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 
 if TYPE_CHECKING:
     from pytcp.lib.socket import Socket
@@ -119,7 +119,7 @@ def cli(*, interface: str) -> None:
     Run the TCP Daytime service.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     service = TcpDaytimeService()
 
     try:

examples/tcp_discard_service.py

@@ -41,7 +41,7 @@
 import click
 
 from examples.lib.tcp_service import TcpService
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 
 if TYPE_CHECKING:
     from pytcp.lib.socket import Socket
@@ -126,7 +126,7 @@ def cli(*, interface: str) -> None:
     Run the TCP Discard service.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     service = TcpDiscardService()
 
     try:

examples/tcp_echo_client.py

@@ -43,7 +43,7 @@
 
 import click
 
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 from pytcp.lib import socket
 from pytcp.lib.ip_helper import ip_version
 
@@ -181,7 +181,7 @@ def cli(*, interface: str, remote_ip_address: str) -> None:
     Run the TCP Echo client.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     client = TcpEchoClient(
         remote_ip_address=remote_ip_address,
     )

examples/tcp_echo_service.py

@@ -42,7 +42,7 @@
 
 from examples.lib.malpi import malpa, malpi, malpka
 from examples.lib.tcp_service import TcpService
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 
 if TYPE_CHECKING:
     from pytcp.lib.socket import Socket
@@ -137,7 +137,7 @@ def cli(*, interface: str) -> None:
     Run the TCP Echo service.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     service = TcpEchoService()
 
     try:

examples/udp_daytime_service.py

@@ -42,7 +42,7 @@
 import click
 
 from examples.lib.udp_service import UdpService
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 
 if TYPE_CHECKING:
     from pytcp.lib.socket import Socket
@@ -89,7 +89,7 @@ def cli(*, interface: str) -> None:
     Run the UDP Daytime service.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     service = UdpDaytimeService()
 
     try:

examples/udp_discard_service.py

@@ -41,7 +41,7 @@
 import click
 
 from examples.lib.udp_service import UdpService
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 
 if TYPE_CHECKING:
     from pytcp.lib.socket import Socket
@@ -86,7 +86,7 @@ def cli(*, interface: str) -> None:
     Run the UDP Discard service.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     service = UdpDiscardService()
 
     try:

examples/udp_echo_client.py

@@ -43,7 +43,7 @@
 
 import click
 
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 from pytcp.lib import socket
 from pytcp.lib.ip_helper import ip_version
 
@@ -176,7 +176,7 @@ def cli(*, interface: str, remote_ip_address: str) -> None:
     Run the TCP Echo client.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     client = UdpEchoClient(
         remote_ip_address=remote_ip_address,
     )

examples/udp_echo_service.py

@@ -42,7 +42,7 @@
 
 from examples.lib.malpi import malpa, malpi, malpka
 from examples.lib.udp_service import UdpService
-from pytcp import TcpIpStack
+from pytcp import TcpIpStack, initialize_tap
 
 if TYPE_CHECKING:
     from pytcp.lib.socket import Socket
@@ -102,7 +102,7 @@ def cli(*, interface: str) -> None:
     Run the UDP Echo service.
     """
 
-    stack = TcpIpStack(interface=interface)
+    stack = TcpIpStack(fd=initialize_tap(tap_name=interface))
     service = UdpEchoService()
 
     try:

pytcp/__init__.py

@@ -37,7 +37,6 @@
 import os
 import struct
 import sys
-from typing import Tuple
 
 from pytcp import config
 from pytcp.lib import stack
@@ -51,35 +50,41 @@
 IFF_NO_PI = 0x1000
 
 
+def initialize_tap(*, tap_name: str) -> tuple[int, int]:
+    """
+    Initialize the TAP interface.
+    """
+
+    try:
+        fd = os.open("/dev/net/tun", os.O_RDWR)
+
+    except FileNotFoundError:
+        log("stack", "<CRIT>Unable to access '/dev/net/tun' device</>")
+        sys.exit(-1)
+
+    fcntl.ioctl(
+        fd,
+        TUNSETIFF,
+        struct.pack("16sH", tap_name.encode(), IFF_TAP | IFF_NO_PI),
+    )
+
+    return fd, fd
+
+
 class TcpIpStack:
     """
     Main PyTCP library class.
     """
-    @staticmethod
-    def create_tun(interface:str):
-        # Initialize the TAP interface.
-        try:
-            fd = os.open("/dev/net/tun", os.O_RDWR)
-        except FileNotFoundError:
-            log("stack", "<CRIT>Unable to access '/dev/net/tun' device</>")
-            sys.exit(-1)
-        fcntl.ioctl(
-            fd,
-            TUNSETIFF,
-            struct.pack("16sH", interface.encode(), IFF_TAP | IFF_NO_PI),
-        )
-        return fd, fd
 
     def __init__(
         self,
         *,
-        fd:Tuple[int,int],
+        fd: tuple[int, int],
         mac_address: str | None = None,
         ip4_address: str | None = None,
         ip4_gateway: str | None = None,
         ip6_address: str | None = None,
         ip6_gateway: str | None = None,
-
     ):
         """
         Initialize stack on given interface.
@@ -119,7 +124,6 @@ def __init__(
         self.rx_fd = fd[0]
         self.tx_fd = fd[1]
 
-
     def start(self) -> None:
         """
         Start stack components.

pytcp/protocols/tcp/session.py

@@ -46,8 +46,9 @@
 
 import random
 import threading
+from collections.abc import Callable
 from enum import Enum, auto
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 
 from pytcp import config
 from pytcp.lib import stack

pytcp/subsystems/rx_ring.py

@@ -41,7 +41,7 @@
 import select
 import threading
 import time
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from pytcp.lib.logger import log
 from pytcp.lib.packet import PacketRx
@@ -107,7 +107,7 @@ def __thread_receive(self) -> None:
 
         __debug__ and log("stack", "Stopped RX ring")
 
-    def dequeue(self) -> Optional[PacketRx]:
+    def dequeue(self) -> PacketRx | None:
         """
         Dequeue inboutd frame from RX ring.
         """

pytcp/subsystems/timer.py

@@ -40,7 +40,8 @@
 
 import threading
 import time
-from typing import Any, Callable
+from collections.abc import Callable
+from typing import Any
 
 from pytcp.lib.logger import log
 

requirements_dev.txt

@@ -10,3 +10,4 @@ pyre-check
 build
 twine
 ipython
+pyupgrade