Fix WHOIS timeout (#4)
Browse files- feat: Prefer whois system binary + increase cache sizes (392e9fb06768bf5b8161a9e503ce631a67ebc0a6)
- tdagent/tools/virus_total.py +6 -1
- tdagent/tools/whois.py +15 -1
tdagent/tools/virus_total.py
CHANGED
|
@@ -6,11 +6,16 @@ import gradio as gr
|
|
| 6 |
import vt
|
| 7 |
|
| 8 |
|
|
|
|
|
|
|
|
|
|
| 9 |
# Get API key from environment variable
|
| 10 |
API_KEY = os.getenv("VT_API_KEY")
|
| 11 |
|
| 12 |
|
| 13 |
-
@cachetools.cached(
|
|
|
|
|
|
|
| 14 |
def get_virus_total_url_info(url: str) -> str:
|
| 15 |
"""Get URL Info from VirusTotal URL Scanner. Scan URL is not available."""
|
| 16 |
if not API_KEY:
|
|
|
|
| 6 |
import vt
|
| 7 |
|
| 8 |
|
| 9 |
+
_CACHE_MAX_SIZE = 4096
|
| 10 |
+
_CACHE_TTL_SECONDS = 3600
|
| 11 |
+
|
| 12 |
# Get API key from environment variable
|
| 13 |
API_KEY = os.getenv("VT_API_KEY")
|
| 14 |
|
| 15 |
|
| 16 |
+
@cachetools.cached(
|
| 17 |
+
cache=cachetools.TTLCache(maxsize=_CACHE_MAX_SIZE, ttl=_CACHE_TTL_SECONDS),
|
| 18 |
+
)
|
| 19 |
def get_virus_total_url_info(url: str) -> str:
|
| 20 |
"""Get URL Info from VirusTotal URL Scanner. Scan URL is not available."""
|
| 21 |
if not API_KEY:
|
tdagent/tools/whois.py
CHANGED
|
@@ -1,11 +1,21 @@
|
|
| 1 |
import json
|
|
|
|
| 2 |
|
|
|
|
| 3 |
import gradio as gr
|
| 4 |
import whois
|
| 5 |
|
| 6 |
from tdagent.utils.json_utils import TDAgentJsonEncoder
|
| 7 |
|
| 8 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 9 |
def query_whois(url: str) -> str:
|
| 10 |
"""Query a WHOIS database to gather information about a url or domain.
|
| 11 |
|
|
@@ -19,7 +29,11 @@ def query_whois(url: str) -> str:
|
|
| 19 |
A JSON formatted string with the gathered information
|
| 20 |
"""
|
| 21 |
try:
|
| 22 |
-
whois_result = whois.whois(
|
|
|
|
|
|
|
|
|
|
|
|
|
| 23 |
except whois.parser.PywhoisError as err:
|
| 24 |
return json.dumps({"error": str(err)})
|
| 25 |
|
|
|
|
| 1 |
import json
|
| 2 |
+
import shutil
|
| 3 |
|
| 4 |
+
import cachetools
|
| 5 |
import gradio as gr
|
| 6 |
import whois
|
| 7 |
|
| 8 |
from tdagent.utils.json_utils import TDAgentJsonEncoder
|
| 9 |
|
| 10 |
|
| 11 |
+
_WHOIS_BINARY = "whois"
|
| 12 |
+
_CACHE_MAX_SIZE = 4096
|
| 13 |
+
_CACHE_TTL_SECONDS = 3600
|
| 14 |
+
|
| 15 |
+
|
| 16 |
+
@cachetools.cached(
|
| 17 |
+
cache=cachetools.TTLCache(maxsize=_CACHE_MAX_SIZE, ttl=_CACHE_TTL_SECONDS),
|
| 18 |
+
)
|
| 19 |
def query_whois(url: str) -> str:
|
| 20 |
"""Query a WHOIS database to gather information about a url or domain.
|
| 21 |
|
|
|
|
| 29 |
A JSON formatted string with the gathered information
|
| 30 |
"""
|
| 31 |
try:
|
| 32 |
+
whois_result = whois.whois(
|
| 33 |
+
url,
|
| 34 |
+
command=shutil.which(_WHOIS_BINARY) is not None,
|
| 35 |
+
executable=_WHOIS_BINARY,
|
| 36 |
+
)
|
| 37 |
except whois.parser.PywhoisError as err:
|
| 38 |
return json.dumps({"error": str(err)})
|
| 39 |
|