diff --git a/.coveragerc b/.coveragerc index d4008ef..a624f92 100644 --- a/.coveragerc +++ b/.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 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ff4ada2..9719026 100644 --- a/.github/workflows/test.yml +++ b/.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: | diff --git a/CHANGELOG.md b/CHANGELOG.md index ceda311..dcb3eeb 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 02ad599..6b286e8 100644 --- a/README.md +++ b/README.md @@ -8,10 +8,10 @@ Discord Channel -## Overview +## Overview

-IPForce is a Python library for TODO. +IPForce 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.

@@ -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 diff --git a/security.md b/SECURITY.md similarity index 100% rename from security.md rename to SECURITY.md diff --git a/ipforce/__init__.py b/ipforce/__init__.py index d701c1d..64e4c5a 100644 --- a/ipforce/__init__.py +++ b/ipforce/__init__.py @@ -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 diff --git a/ipforce/adapters.py b/ipforce/adapters.py index 326520f..4e3e828 100644 --- a/ipforce/adapters.py +++ b/ipforce/adapters.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 diff --git a/ipforce/params.py b/ipforce/params.py index ce1161f..401b71f 100644 --- a/ipforce/params.py +++ b/ipforce/params.py @@ -4,4 +4,3 @@ IPFORCE_VERSION = "0.1" IPFORCE_OVERVIEW = '''OVERVIEW''' IPFORCE_REPO = "https://github.com/openscilab/ipforce" - diff --git a/tests/test_adapters.py b/tests/test_adapters.py new file mode 100644 index 0000000..659da30 --- /dev/null +++ b/tests/test_adapters.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) diff --git a/tests/test_ipv4.py b/tests/test_ipv4.py new file mode 100644 index 0000000..9b740d7 --- /dev/null +++ b/tests/test_ipv4.py @@ -0,0 +1,33 @@ +import requests +import ipaddress +from ipforce import IPv4TransportAdapter + + +def is_ipv4(ip: str) -> bool: + """ + Check if the given input is a valid IPv4 address. + + :param ip: input IP + """ + if not isinstance(ip, str): + return False + try: + _ = ipaddress.IPv4Address(ip) + return True + except Exception: + return False + + +def test_ipv4_adapter(): + """Test the IPv4 adapter by making a request that will only use IPv4 addresses.""" + print("Testing IPv4 Adapter...") + + # Create a session with IPv4 adapter + with requests.Session() as session: + ipv4Addapter = IPv4TransportAdapter() + session.mount('http://', ipv4Addapter) + session.mount('https://', ipv4Addapter) + # This will only resolve to IPv4 addresses + response = session.get('https://ifconfig.co/json', timeout=10) + response.raise_for_status() + assert is_ipv4(response.json()['ip']) diff --git a/tests/test_ipv6.py b/tests/test_ipv6.py new file mode 100644 index 0000000..5289049 --- /dev/null +++ b/tests/test_ipv6.py @@ -0,0 +1,33 @@ +import requests +import ipaddress +from ipforce import IPv6TransportAdapter + + +def is_ipv6(ip: str) -> bool: + """ + Check if the given input is a valid IPv6 address. + + :param ip: input IP + """ + if not isinstance(ip, str): + return False + try: + _ = ipaddress.IPv6Address(ip) + return True + except Exception: + return False + + +def test_ipv6_adapter(): + """Test the IPv6 adapter by making a request that will only use IPv6 addresses.""" + print("\nTesting IPv6 Adapter...") + + # Create a session with IPv6 adapter + with requests.Session() as session: + ipv6Addapter = IPv6TransportAdapter() + session.mount('http://', ipv6Addapter) + session.mount('https://', ipv6Addapter) + # This will only resolve to IPv6 addresses + response = session.get('https://ifconfig.co/json', timeout=10) + response.raise_for_status() + assert is_ipv6(response.json()['ip'])