Merge pull request #1 from openscilab/add/adapters

Add/adapters

Author: AHReccese

.coveragerc

@@ -1,8 +1,6 @@
 [run]
 branch = True
 omit =
-    */ipforce/cli.py
-    */ipforce/__main__.py
     */ipforce/__init__.py
 [report]
 # Regexes for lines to exclude from consideration

.github/workflows/test.yml

@@ -24,12 +24,12 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-22.04, windows-2022, macOS-13]
-        python-version: [3.7, 3.8, 3.9, 3.10.5, 3.11.0, 3.12.0, 3.13.0]
+        os: [ubuntu-22.04, windows-2022, macos-15-intel]
+        python-version: [3.7, 3.8, 3.9, 3.10.5, 3.11.0, 3.12.0, 3.13.0, 3.14.0]
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v5
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
     - name: Installation
@@ -42,12 +42,11 @@ jobs:
         pip install --upgrade --upgrade-strategy=only-if-needed -r test-requirements.txt
     - name: Test with pytest 
       run: |
-        python -m pytest . --cov=ipforce --cov-report=term
+        python -m pytest tests/test_adapters.py --cov=ipforce --cov-report=term
     - name: Upload coverage to Codecov
       uses: codecov/codecov-action@v4
       with:
-        fail_ci_if_error: true
-        token: ${{ secrets.CODECOV_TOKEN }}
+        fail_ci_if_error: false
       if: matrix.python-version == env.TEST_PYTHON_VERSION && matrix.os == env.TEST_OS
     - name: Version check
       run: |

CHANGELOG.md

@@ -5,7 +5,11 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+## [0.1] - 2025-xx-xx
 ### Added
-### Changed
+- Test system
+- IPv6TransportAdapter
+- IPv4TransportAdapter
 
-[Unreleased]: https://github.com/openscilab/ipforce/compare/TODO...v0.1
+[Unreleased]: https://github.com/openscilab/ipforce/compare/v0.1...dev
+[0.1]: https://github.com/openscilab/ipforce/compare/7128b04...v0.1

README.md

@@ -8,10 +8,10 @@
 <a href="https://discord.gg/TODO"><img src="https://img.shields.io/discord/TODO" alt="Discord Channel"></a>
 </div>			
 				
-## Overview	
+## Overview
 
 <p align="justify">					
-<b>IPForce</b> is a Python library for TODO.
+<b>IPForce</b> is a Python library that provides HTTP adapters for forcing specific IP protocol versions (IPv4 or IPv6) during HTTP requests. It's particularly useful for testing network connectivity, ensuring compatibility with specific network configurations, and controlling which IP protocol version is used for DNS resolution and connections.
 </p>
 
 <table>
@@ -55,16 +55,46 @@
 - `pip install .`				
 
 ### PyPI
-
 - Check [Python Packaging User Guide](https://packaging.python.org/installing/)     
 - `pip install ipforce==0.1`						
 
-
 ## Usage
+### Enforce IPv4
+
+Use when you need to ensure connections only use IPv4 addresses, useful for legacy systems that don't support IPv6, networks with IPv4-only infrastructure, or testing IPv4 connectivity.
+
+```python
+import requests
+from ipforce import IPv4TransportAdapter
+
+# Create a session that will only use IPv4 addresses
+session = requests.Session()
+session.mount('http://', IPv4TransportAdapter())
+session.mount('https://', IPv4TransportAdapter())
+
+# All requests through this session will only resolve to IPv4 addresses
+response = session.get('https://ifconfig.co/json')
+```
+
+### Enforce IPv6
+
+Use when you need to ensure connections only use IPv6 addresses, useful for modern networks with IPv6 infrastructure, testing IPv6 connectivity, or applications requiring IPv6-specific features.
+
+```python
+import requests
+from ipforce import IPv6TransportAdapter
+
+# Create a session that will only use IPv6 addresses
+session = requests.Session()
+session.mount('http://', IPv6TransportAdapter())
+session.mount('https://', IPv6TransportAdapter())
 
-### Library
+# All requests through this session will only resolve to IPv6 addresses
+response = session.get('https://ifconfig.co/json')
+```
 
-#### Enforce IPv4
+> [!WARNING]
+> Current adapters are NOT thread-safe! They modify the global `socket.getaddrinfo` function, which can cause issues in multi-threaded applications.
 
 ## Issues & Bug Reports			
 

security.md SECURITY.mdsecurity.md renamed to SECURITY.md

@@ -1,5 +1,6 @@
+# -*- coding: utf-8 -*-
 """ipforce modules."""
 from .params import IPFORCE_VERSION
-from .adapters import IPv4HTTPAdapter
+from .adapters import IPv4TransportAdapter, IPv6TransportAdapter
 
 __version__ = IPFORCE_VERSION

ipforce/__init__.py

@@ -1,61 +1,65 @@
+# -*- coding: utf-8 -*-
+"""IPForce Adapters to force IPv4 or IPv6 for requests."""
 import socket
-from typing import List, Tuple
-from urllib3 import PoolManager
+from typing import Any, List, Tuple
 from requests.adapters import HTTPAdapter
-from requests.sessions import Session
 
-class IPv4HTTPAdapter(HTTPAdapter):
+
+class IPv4TransportAdapter(HTTPAdapter):
     """A custom HTTPAdapter that enforces the use of IPv4 for DNS resolution during HTTP(S) requests using the requests library."""
 
-    def init_poolmanager(self, connections: int, maxsize: int, block: bool = False, **kwargs: dict) -> None:
-        """
-        Initialize the connection pool manager using a temporary override of socket.getaddrinfo to ensure only IPv4 addresses are used.
-        This is necessary to ensure that the requests library uses IPv4 addresses for DNS resolution, which is required for some APIs.
-        :param connections: the number of connection pools to cache
-        :param maxsize: the maximum number of connections to save in the pool
-        :param block: whether the connections should block when reaching the max size
-        :param kwargs: additional keyword arguments for the PoolManager
+    def send(self, *args: list, **kwargs: dict) -> Any:
         """
-        self.poolmanager = PoolManager(
-            num_pools=connections,
-            maxsize=maxsize,
-            block=block,
-            socket_options=self._ipv4_socket_options(),
-            **kwargs
-        )
-
-    def _ipv4_socket_options(self) -> list:
-        """
-        Temporarily patches socket.getaddrinfo to filter only IPv4 addresses (AF_INET).
+        Override send method to apply the monkey patch only during the request.
 
-        :return: an empty list of socket options; DNS patching occurs here
+        :param args: additional list arguments for the send method
+        :param kwargs: additional keyword arguments for the send method
         """
         original_getaddrinfo = socket.getaddrinfo
 
-        def ipv4_only_getaddrinfo(*args: list, **kwargs: dict) -> List[Tuple]:
-            results = original_getaddrinfo(*args, **kwargs)
+        def ipv4_only_getaddrinfo(*gargs: list, **gkwargs: dict) -> List[Tuple]:
+            """
+            Filter getaddrinfo to return only IPv4 addresses.
+
+            :param gargs: additional list arguments for the original_getaddrinfo function
+            :param gkwargs: additional keyword arguments for the original_getaddrinfo function
+            """
+            results = original_getaddrinfo(*gargs, **gkwargs)
             return [res for res in results if res[0] == socket.AF_INET]
 
-        self._original_getaddrinfo = socket.getaddrinfo
         socket.getaddrinfo = ipv4_only_getaddrinfo
+        try:
+            response = super().send(*args, **kwargs)
+        finally:
+            socket.getaddrinfo = original_getaddrinfo
+        return response
 
-        return []
 
-    def __del__(self) -> None:
-        """Restores the original socket.getaddrinfo function upon adapter deletion."""
-        if hasattr(self, "_original_getaddrinfo"):
-            socket.getaddrinfo = self._original_getaddrinfo
+class IPv6TransportAdapter(HTTPAdapter):
+    """A custom HTTPAdapter that enforces the use of IPv6 for DNS resolution during HTTP(S) requests using the requests library."""
 
-    @staticmethod
-    def get_ipv4_enforced_session() -> Session:
+    def send(self, *args: list, **kwargs: dict) -> Any:
         """
-        Returns a requests.Session with IPv4HTTPAdapter mounted for both HTTP and HTTPS.
-        All requests made with this session will use IPv4 for DNS resolution.
+        Override send method to apply the monkey patch only during the request.
 
-        :return: requests.Session object with IPv4 enforced
+        :param args: additional list arguments for the send method
+        :param kwargs: additional keyword arguments for the send method
         """
-        session = Session()
-        adapter = IPv4HTTPAdapter()
-        session.mount("http://", adapter)
-        session.mount("https://", adapter)
-        return session
+        original_getaddrinfo = socket.getaddrinfo
+
+        def ipv6_only_getaddrinfo(*gargs: list, **gkwargs: dict) -> List[Tuple]:
+            """
+            Filter getaddrinfo to return only IPv6 addresses.
+
+            :param gargs: additional list arguments for the original_getaddrinfo function
+            :param gkwargs: additional keyword arguments for the original_getaddrinfo function
+            """
+            results = original_getaddrinfo(*gargs, **gkwargs)
+            return [res for res in results if res[0] == socket.AF_INET6]
+
+        socket.getaddrinfo = ipv6_only_getaddrinfo
+        try:
+            response = super().send(*args, **kwargs)
+        finally:
+            socket.getaddrinfo = original_getaddrinfo
+        return response

ipforce/adapters.py

@@ -4,4 +4,3 @@
 IPFORCE_VERSION = "0.1"
 IPFORCE_OVERVIEW = '''OVERVIEW'''
 IPFORCE_REPO = "https://github.com/openscilab/ipforce"
-

ipforce/params.py

@@ -0,0 +1,149 @@
+import unittest
+import socket
+from unittest.mock import patch, MagicMock
+from ipforce.adapters import IPv4TransportAdapter, IPv6TransportAdapter
+
+
+class TestIPv4Adapter(unittest.TestCase):
+    """Test cases for IPv4TransportAdapter."""
+
+    def setup(self):
+        """Set up test fixtures."""
+        self.adapter = IPv4TransportAdapter()
+
+    def test_ipv4_filtering_during_send(self):
+        """Test that IPv4 adapter filters only IPv4 addresses during send."""
+        self.setup()
+        mock_results = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('192.168.1.1', 80)),  # IPv4
+            (socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('::1', 80)),         # IPv6
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('10.0.0.1', 80)),     # IPv4
+        ]
+
+        original_getaddrinfo = socket.getaddrinfo
+        captured_results = []
+
+        def mock_super_send(*args, **kwargs):
+            # Capture the filtered results during send
+            captured_results.extend(socket.getaddrinfo('example.com', 80))
+            return MagicMock()
+
+        with patch('socket.getaddrinfo', return_value=mock_results):
+            with patch.object(IPv4TransportAdapter.__bases__[0], 'send', mock_super_send):
+                self.adapter.send(MagicMock())
+
+        # Only IPv4 results should be captured
+        self.assertEqual(len(captured_results), 2)
+        for result in captured_results:
+            self.assertEqual(result[0], socket.AF_INET)
+
+    def test_cleanup_after_send(self):
+        """Test that the adapter properly restores original getaddrinfo after send."""
+        self.setup()
+        original_getaddrinfo = socket.getaddrinfo
+
+        with patch.object(IPv4TransportAdapter.__bases__[0], 'send', return_value=MagicMock()):
+            self.adapter.send(MagicMock())
+
+        # Verify it was restored after send
+        self.assertEqual(socket.getaddrinfo, original_getaddrinfo)
+
+    def test_cleanup_on_exception(self):
+        """Test that the adapter restores original getaddrinfo even if send raises."""
+        self.setup()
+        original_getaddrinfo = socket.getaddrinfo
+
+        with patch.object(IPv4TransportAdapter.__bases__[0], 'send', side_effect=Exception("Test error")):
+            with self.assertRaises(Exception):
+                self.adapter.send(MagicMock())
+
+        # Verify it was restored even after exception
+        self.assertEqual(socket.getaddrinfo, original_getaddrinfo)
+
+
+class TestIPv6Adapter(unittest.TestCase):
+    """Test cases for IPv6TransportAdapter."""
+
+    def setup(self):
+        """Set up test fixtures."""
+        self.adapter = IPv6TransportAdapter()
+
+    def test_ipv6_filtering_during_send(self):
+        """Test that IPv6 adapter filters only IPv6 addresses during send."""
+        self.setup()
+        mock_results = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, '', ('192.168.1.1', 80)),  # IPv4
+            (socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('::1', 80)),         # IPv6
+            (socket.AF_INET6, socket.SOCK_STREAM, 6, '', ('2001:db8::1', 80)), # IPv6
+        ]
+
+        captured_results = []
+
+        def mock_super_send(*args, **kwargs):
+            # Capture the filtered results during send
+            captured_results.extend(socket.getaddrinfo('example.com', 80))
+            return MagicMock()
+
+        with patch('socket.getaddrinfo', return_value=mock_results):
+            with patch.object(IPv6TransportAdapter.__bases__[0], 'send', mock_super_send):
+                self.adapter.send(MagicMock())
+
+        # Only IPv6 results should be captured
+        self.assertEqual(len(captured_results), 2)
+        for result in captured_results:
+            self.assertEqual(result[0], socket.AF_INET6)
+
+    def test_cleanup_after_send(self):
+        """Test that the adapter properly restores original getaddrinfo after send."""
+        self.setup()
+        original_getaddrinfo = socket.getaddrinfo
+
+        with patch.object(IPv6TransportAdapter.__bases__[0], 'send', return_value=MagicMock()):
+            self.adapter.send(MagicMock())
+
+        # Verify it was restored after send
+        self.assertEqual(socket.getaddrinfo, original_getaddrinfo)
+
+    def test_cleanup_on_exception(self):
+        """Test that the adapter restores original getaddrinfo even if send raises."""
+        self.setup()
+        original_getaddrinfo = socket.getaddrinfo
+
+        with patch.object(IPv6TransportAdapter.__bases__[0], 'send', side_effect=Exception("Test error")):
+            with self.assertRaises(Exception):
+                self.adapter.send(MagicMock())
+
+        # Verify it was restored even after exception
+        self.assertEqual(socket.getaddrinfo, original_getaddrinfo)
+
+
+class TestAdapterIntegration(unittest.TestCase):
+    """Integration tests for both adapters."""
+
+    def test_both_adapters_independent(self):
+        """Test that both adapters can coexist without interference."""
+        ipv4_adapter = IPv4TransportAdapter()
+        ipv6_adapter = IPv6TransportAdapter()
+        original_getaddrinfo = socket.getaddrinfo
+
+        with patch.object(IPv4TransportAdapter.__bases__[0], 'send', return_value=MagicMock()):
+            ipv4_adapter.send(MagicMock())
+
+        with patch.object(IPv6TransportAdapter.__bases__[0], 'send', return_value=MagicMock()):
+            ipv6_adapter.send(MagicMock())
+
+        # Verify original is still intact
+        self.assertEqual(socket.getaddrinfo, original_getaddrinfo)
+
+    def test_sequential_sends_restore_correctly(self):
+        """Test that multiple sequential sends properly restore getaddrinfo."""
+        adapter = IPv4TransportAdapter()
+        original_getaddrinfo = socket.getaddrinfo
+
+        with patch.object(IPv4TransportAdapter.__bases__[0], 'send', return_value=MagicMock()):
+            adapter.send(MagicMock())
+            adapter.send(MagicMock())
+            adapter.send(MagicMock())
+
+        # Verify original is still intact after multiple sends
+        self.assertEqual(socket.getaddrinfo, original_getaddrinfo)