gatenet 0.9.5

Python library for gateways, networks, and devices.

gatenet 🛰️

BETA

Package
Python
Tests
License

Gatenet is a batteries-included Python networking toolkit for diagnostics, service discovery, and building robust socket, UDP, and HTTP microservices.

• Diagnostics: Traceroute, ping/latency, bandwidth, port scanning, geo IP, and more.
• Service Discovery: Identify running services (SSH, HTTP, FTP, SMTP, mDNS, SSDP, Bluetooth, etc.) using banners, ports, and extensible detectors.
• Socket & HTTP: Modular TCP/UDP/HTTP servers and clients, with async support.
• Extensible: Strategy and chain-of-responsibility patterns for easy extension and custom detection.
• Comprehensive tests and documentation.

Gatenet is designed for developers who need reliable, extensible, and well-tested networking tools for diagnostics, automation, and microservice development.

• Changelog
• Installation
• Features
• Quickstart
• Usage Examples
• Service Discovery
• Diagnostics
• Tests
• Contributing
• License

---

Installation

pip install gatenet

Features

• Modular: Each component is modular and can be used independently.
• Testable: Each component is designed to be testable with unit tests.
• Service Discovery: Identify running services (SSH, HTTP, FTP, SMTP, etc.) using banners and ports.
• Diagnostics: Tools for traceroute, latency, and bandwidth measurement.
• Socket Servers & Clients: TCP, UDP, and HTTP server/client implementations.
• Async Support: Asynchronous HTTP client and server.
• Extensible: Strategy and chain-of-responsibility patterns for easy extension.
• Comprehensive Documentation: With examples and usage guides.

---

Quickstart

TCP Client

# Synchronous usage
from gatenet.client.tcp import TCPClient

client = TCPClient(host="127.0.0.1", port=12345)
client.connect()
response = client.send("ping")
print(response)
client.close()

# --- Async usage ---
from gatenet.client.tcp import AsyncTCPClient
import asyncio

async def main():
    client = AsyncTCPClient(host="127.0.0.1", port=12345)
    await client.connect()
    response = await client.send("ping")
    print(response)
    await client.close()

asyncio.run(main())

HTTP Server

# Synchronous usage
from gatenet.http.server import HTTPServer

server = HTTPServer(host="0.0.0.0", port=8080)

@server.route("/status", method="GET")
def status_handler(req):
    return {"ok": True}

server.serve()

# --- Async usage ---
from gatenet.http_.async_client import AsyncHTTPClient
import asyncio

async def main():
    client = AsyncHTTPClient("http://localhost:8080")
    response = await client.get("/status")
    print(response)

asyncio.run(main())

---

Usage Examples

Service Discovery

Identify a service by port and banner:

from gatenet.discovery.ssh import _identify_service

service = _identify_service(22, "SSH-2.0-OpenSSH_8.9p1")
print(service)  # Output: "OpenSSH 8.9p1"

Use a specific detector:

from gatenet.discovery.ssh import HTTPDetector

detector = HTTPDetector()
result = detector.detect(80, "apache/2.4.41")
print(result)  # Output: "Apache HTTP Server"

Custom Service Detector

from gatenet.discovery.ssh import ServiceDetector
from typing import Optional

class CustomDetector(ServiceDetector):
    """Custom service detector implementation."""
    def detect(self, port: int, banner: str) -> Optional[str]:
        if 'myapp' in banner:
            return "MyCustomApp"
        return None

---

Service Discovery

• SSH, HTTP, FTP, SMTP, and more: Uses a strategy pattern and chain of responsibility for extensible service detection.
• Banner and port-based detection: Extracts service names and versions from banners and well-known ports.
• Fallback detection: Always returns a result, even for unknown services.

See examples/discovery/ssh_discovery.py for more.

---

Diagnostics

• Traceroute: Trace the route to a host.
• Latency Measurement: Measure round-trip time to a host.
• Bandwidth Measurement: Measure throughput to a host.

Example traceroute:

from gatenet.diagnostics.traceroute import traceroute

hops = traceroute("google.com")
for hop in hops:
    print(hop)

Example bandwidth measurement:

from gatenet.diagnostics.bandwidth import measure_bandwidth

result = measure_bandwidth("google.com")
print(f"Download: {result['download_mbps']} Mbps, Upload: {result['upload_mbps']} Mbps")

---

Tests

Run all tests with:

pytest

• Uses pytest for all tests.
• Includes unit and integration tests for all modules.
• Use get_free_port() from gatenet.utils.net in tests to avoid port conflicts.

---

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

• Follow the code style and patterns used in the project.
• Add tests for new features.
• Update documentation as needed.

---

License

MIT