Skip to content

fix: share one HTTP failure helper (Closes #485) #486

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Cid-oe wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: share one HTTP failure helper (Closes #485) #486

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 3 additions & 13 deletions modules/cert_transparency.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,17 +4,7 @@
sys.path.append('..')
from config import Colors, USER_AGENT
from modules import get_proxies, get_with_retry
from modules.module_status import annotate, classify, reason_for, OK, RATE_LIMITED, ERROR


def _http_failure(result: Dict[str, Any], source: str, status_code: int) -> Dict[str, Any]:
if status_code in (429, 503):
return annotate(
result,
RATE_LIMITED,
f"{source} is busy or rate limited (HTTP {status_code}); try again later",
)
return annotate(result, ERROR, f"{source} returned status {status_code}")
from modules.module_status import annotate, classify, reason_for, OK, RATE_LIMITED, ERROR, annotate_http_failure


class CertTransparency:
Expand Down Expand Up @@ -44,7 +34,7 @@ def _search_crtsh(self, domain: str) -> Dict[str, Any]:


if response.status_code != 200:
return _http_failure(result, "crt.sh", response.status_code)
return annotate_http_failure(result, "crt.sh", response.status_code)

try:
certs = response.json()
Expand Down Expand Up @@ -144,7 +134,7 @@ def _search_certspotter(self, domain: str) -> Dict[str, Any]:
proxies=get_proxies(),
)
if response.status_code != 200:
return _http_failure(result, "certspotter", response.status_code)
return annotate_http_failure(result, "certspotter", response.status_code)
try:
issuances = response.json()
except Exception:
Expand Down
9 changes: 9 additions & 0 deletions modules/module_status.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,3 +122,12 @@ def annotate(result: Dict[str, Any], status: str, reason: Optional[str] = None)
else:
result["error"] = None
return result

def annotate_http_failure(result: Dict[str, Any], source: str, status_code: int) -> Dict[str, Any]:
if status_code in (429, 503):
return annotate(
result,
RATE_LIMITED,
f"{source} is busy or rate limited (HTTP {status_code}); try again later",
)
return annotate(result, ERROR, f"{source} returned status {status_code}")
18 changes: 4 additions & 14 deletions modules/wayback.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,7 @@
sys.path.append('..')
from config import Colors
from modules import get_proxies, get_with_retry
from modules.module_status import annotate, OK, RATE_LIMITED, ERROR


def _http_failure(result: Dict[str, Any], source: str, status_code: int) -> Dict[str, Any]:
if status_code in (429, 503):
return annotate(
result,
RATE_LIMITED,
f"{source} is busy or rate limited (HTTP {status_code}); try again later",
)
return annotate(result, ERROR, f"{source} returned status {status_code}")
from modules.module_status import annotate, OK, RATE_LIMITED, ERROR, annotateannotate_http_failure


class WaybackMachine:
Expand Down Expand Up @@ -51,7 +41,7 @@ def get_snapshots(self, url: str, limit: int = 20) -> Dict[str, Any]:
)

if r.status_code != 200:
return _http_failure(result, "CDX API", r.status_code)
return annotate_http_failure(result, "CDX API", r.status_code)

rows = r.json()
if not rows or len(rows) < 2:
Expand Down Expand Up @@ -131,7 +121,7 @@ def get_all_urls(self, domain: str, limit: int = 100) -> Dict[str, Any]:
)

if r.status_code != 200:
return _http_failure(result, "CDX API", r.status_code)
return annotate_http_failure(result, "CDX API", r.status_code)

rows = r.json()
if not rows or len(rows) < 2:
Expand Down Expand Up @@ -170,7 +160,7 @@ def check_availability(self, url: str) -> Dict[str, Any]:
result["closest_snapshot"] = closest.get("url")
return annotate(result, OK)
else:
return _http_failure(result, "Wayback availability API", r.status_code)
return annotate_http_failure(result, "Wayback availability API", r.status_code)
except Exception as e:
return annotate(result, ERROR, str(e))

Expand Down
17 changes: 17 additions & 0 deletions tests/test_module_status.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,3 +189,20 @@ def test_telegram_id_skipped_without_token(self):
assert result["error"] is None



class TestAnnotateHttpFailure:
def test_annotate_http_failure_rate_limited(self):
from modules.module_status import annotate_http_failure, RATE_LIMITED
result = annotate_http_failure({}, "test_source", 429)
assert result["status"] == RATE_LIMITED
assert result["status_reason"] == "test_source is busy or rate limited (HTTP 429); try again later"

result = annotate_http_failure({}, "test_source", 503)
assert result["status"] == RATE_LIMITED
assert result["status_reason"] == "test_source is busy or rate limited (HTTP 503); try again later"

def test_annotate_http_failure_error(self):
from modules.module_status import annotate_http_failure, ERROR
result = annotate_http_failure({}, "test_source", 500)
assert result["status"] == ERROR
assert result["status_reason"] == "test_source returned status 500"
Loading