167 lines
4.3 KiB
Python
167 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
import socket
|
|
import time
|
|
import urllib.request
|
|
|
|
from core.tools.base import BaseTool, ToolContext
|
|
from core.tools.registry import registry
|
|
from core.events import bus
|
|
|
|
|
|
class NetTool(BaseTool):
|
|
"""
|
|
Network diagnostics and introspection tool.
|
|
|
|
Provides basic connectivity checks and resolution utilities.
|
|
"""
|
|
|
|
name = "net"
|
|
description = "Network diagnostics: DNS, ports, HTTP checks"
|
|
|
|
# =========================================================
|
|
# EXECUTE
|
|
# =========================================================
|
|
|
|
def execute(self, payload: dict[str, Any], ctx: ToolContext):
|
|
action = str(payload.get("action", "dns")).strip()
|
|
|
|
bus.log(
|
|
"NET",
|
|
"net_execute",
|
|
"INFO",
|
|
{"action": action}
|
|
)
|
|
|
|
match action:
|
|
case "dns":
|
|
return self.dns_lookup(payload)
|
|
|
|
case "port":
|
|
return self.check_port(payload)
|
|
|
|
case "http":
|
|
return self.http_check(payload)
|
|
|
|
case "latency":
|
|
return self.latency_test(payload)
|
|
|
|
case _:
|
|
raise ValueError(f"Unknown net action: {action}")
|
|
|
|
# =========================================================
|
|
# DNS LOOKUP
|
|
# =========================================================
|
|
|
|
def dns_lookup(self, payload: dict[str, Any]):
|
|
host = payload.get("host")
|
|
|
|
if not isinstance(host, str):
|
|
raise ValueError("host must be string")
|
|
|
|
try:
|
|
ip = socket.gethostbyname(host)
|
|
|
|
return {
|
|
"host": host,
|
|
"ip": ip
|
|
}
|
|
|
|
except socket.gaierror as e:
|
|
return {
|
|
"host": host,
|
|
"error": str(e)
|
|
}
|
|
|
|
# =========================================================
|
|
# PORT CHECK
|
|
# =========================================================
|
|
|
|
def check_port(self, payload: dict[str, Any]):
|
|
host = payload.get("host", "127.0.0.1")
|
|
port = payload.get("port")
|
|
|
|
if not isinstance(host, str):
|
|
raise ValueError("host must be string")
|
|
if not isinstance(port, int):
|
|
raise ValueError("port must be int")
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.settimeout(2)
|
|
|
|
try:
|
|
result = sock.connect_ex((host, port))
|
|
return {
|
|
"host": host,
|
|
"port": port,
|
|
"open": result == 0
|
|
}
|
|
finally:
|
|
sock.close()
|
|
|
|
# =========================================================
|
|
# HTTP CHECK
|
|
# =========================================================
|
|
|
|
def http_check(self, payload: dict[str, Any]):
|
|
url = payload.get("url")
|
|
|
|
if not isinstance(url, str):
|
|
raise ValueError("url must be string")
|
|
|
|
try:
|
|
start = time.time()
|
|
|
|
req = urllib.request.Request(
|
|
url,
|
|
method="GET"
|
|
)
|
|
|
|
with urllib.request.urlopen(req, timeout=5) as response:
|
|
status = response.getcode()
|
|
|
|
return {
|
|
"url": url,
|
|
"status_code": status,
|
|
"latency_ms": round((time.time() - start) * 1000, 2)
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"url": url,
|
|
"error": str(e)
|
|
}
|
|
|
|
# =========================================================
|
|
# LATENCY TEST
|
|
# =========================================================
|
|
|
|
def latency_test(self, payload: dict[str, Any]):
|
|
host = payload.get("host")
|
|
|
|
if not isinstance(host, str):
|
|
raise ValueError("host must be string")
|
|
|
|
try:
|
|
start = time.time()
|
|
socket.gethostbyname(host)
|
|
latency = (time.time() - start) * 1000
|
|
|
|
return {
|
|
"host": host,
|
|
"dns_latency_ms": round(latency, 2)
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"host": host,
|
|
"error": str(e)
|
|
}
|
|
|
|
|
|
# =========================================================
|
|
# REGISTER
|
|
# =========================================================
|
|
|
|
registry.register(NetTool()) |
