410 lines
13 KiB
Python
410 lines
13 KiB
Python
"""
|
|
©AngelaMos | 2025
|
|
Authentication and authorization vulnerability scanner
|
|
|
|
OWASP API2:2023
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
from typing import Any
|
|
|
|
from config import settings
|
|
from core.enums import (
|
|
ScanStatus,
|
|
Severity,
|
|
TestType,
|
|
)
|
|
from schemas.test_result_schemas import TestResultCreate
|
|
|
|
from .payloads import AuthPayloads
|
|
from .base_scanner import BaseScanner
|
|
|
|
|
|
class AuthScanner(BaseScanner):
|
|
"""
|
|
Tests for broken authentication vulnerabilities
|
|
|
|
Detects:
|
|
- Missing authentication on endpoints
|
|
- Weak/invalid token acceptance
|
|
- JWT vulnerabilities (none algorithm, weak secrets)
|
|
- Missing rate limiting on auth endpoints
|
|
|
|
Maps to OWASP API Security Top 10 2023: API2:2023
|
|
"""
|
|
def scan(self) -> TestResultCreate:
|
|
"""
|
|
Execute authentication tests
|
|
|
|
Returns:
|
|
TestResultCreate: Scan result with findings
|
|
"""
|
|
missing_auth_test = self._test_missing_authentication()
|
|
if missing_auth_test["vulnerable"]:
|
|
return self._create_vulnerable_result(
|
|
details = "Endpoint accessible without authentication",
|
|
evidence = missing_auth_test,
|
|
severity = Severity.HIGH,
|
|
recommendations = [
|
|
"Require authentication for all sensitive endpoints",
|
|
"Implement proper authentication middleware",
|
|
"Return 401 Unauthorized for missing/invalid credentials",
|
|
],
|
|
)
|
|
|
|
if self.auth_token:
|
|
jwt_test = self._test_jwt_vulnerabilities()
|
|
if jwt_test["vulnerable"]:
|
|
return self._create_vulnerable_result(
|
|
details =
|
|
f"JWT vulnerability: {jwt_test['vulnerability_type']}",
|
|
evidence = jwt_test,
|
|
severity = Severity.CRITICAL,
|
|
recommendations = jwt_test.get(
|
|
"recommendations",
|
|
[
|
|
"Properly validate JWT signatures",
|
|
"Reject 'none' algorithm tokens",
|
|
"Use strong secrets (256+ bits)",
|
|
"Implement token expiration checks",
|
|
],
|
|
),
|
|
)
|
|
|
|
invalid_token_test = self._test_invalid_token_handling()
|
|
if invalid_token_test["vulnerable"]:
|
|
return self._create_vulnerable_result(
|
|
details = "Invalid tokens accepted by endpoint",
|
|
evidence = invalid_token_test,
|
|
severity = Severity.HIGH,
|
|
recommendations = [
|
|
"Reject invalid/malformed tokens with 401 status",
|
|
"Validate token format, signature, and expiration",
|
|
"Log authentication failures for monitoring",
|
|
],
|
|
)
|
|
|
|
return TestResultCreate(
|
|
test_name = TestType.AUTH,
|
|
status = ScanStatus.SAFE,
|
|
severity = Severity.INFO,
|
|
details = "Authentication properly implemented",
|
|
evidence_json = {
|
|
"missing_auth_test": missing_auth_test,
|
|
"invalid_token_test": invalid_token_test,
|
|
},
|
|
recommendations_json = [
|
|
"Authentication is properly configured",
|
|
"Consider implementing additional security measures (2FA, refresh tokens)",
|
|
],
|
|
)
|
|
|
|
def _test_missing_authentication(self) -> dict[str, Any]:
|
|
"""
|
|
Test if endpoint requires authentication
|
|
|
|
Attempts to access endpoint without credentials.
|
|
|
|
Returns:
|
|
dict[str, Any]: Test results
|
|
"""
|
|
session_without_auth = self.session.__class__()
|
|
session_without_auth.headers.update(
|
|
{
|
|
"User-Agent":
|
|
f"{settings.APP_NAME}/{settings.VERSION}",
|
|
"Accept": "application/json",
|
|
}
|
|
)
|
|
|
|
try:
|
|
response = session_without_auth.get(
|
|
self.target_url,
|
|
timeout = settings.SCANNER_CONNECTION_TIMEOUT,
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return {
|
|
"vulnerable":
|
|
True,
|
|
"status_code":
|
|
response.status_code,
|
|
"response_length":
|
|
len(response.text),
|
|
"description":
|
|
"Endpoint accessible without authentication",
|
|
}
|
|
|
|
if response.status_code in (401, 403):
|
|
return {
|
|
"vulnerable":
|
|
False,
|
|
"status_code":
|
|
response.status_code,
|
|
"description":
|
|
"Endpoint properly requires authentication",
|
|
}
|
|
|
|
return {
|
|
"vulnerable": False,
|
|
"status_code": response.status_code,
|
|
"description": "Endpoint returned unexpected status",
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"vulnerable": False,
|
|
"error": str(e),
|
|
"description":
|
|
"Error testing authentication requirement",
|
|
}
|
|
|
|
def _test_jwt_vulnerabilities(self) -> dict[str, Any]:
|
|
"""
|
|
Test for common JWT vulnerabilities
|
|
|
|
Tests:
|
|
- None algorithm acceptance
|
|
- Signature removal
|
|
- Weak secret detection (common patterns)
|
|
|
|
Returns:
|
|
dict[str, Any]: JWT vulnerability test results
|
|
"""
|
|
if not self.auth_token or self.auth_token.count(".") != 2:
|
|
return {
|
|
"vulnerable": False,
|
|
"description": "No valid JWT token provided",
|
|
}
|
|
|
|
none_alg_test = self._test_none_algorithm()
|
|
if none_alg_test["vulnerable"]:
|
|
return none_alg_test
|
|
|
|
signature_removal_test = self._test_signature_removal()
|
|
if signature_removal_test["vulnerable"]:
|
|
return signature_removal_test
|
|
|
|
return {
|
|
"vulnerable": False,
|
|
"tests_performed":
|
|
["none_algorithm",
|
|
"signature_removal"],
|
|
"description": "No JWT vulnerabilities detected",
|
|
}
|
|
|
|
def _test_none_algorithm(self) -> dict[str, Any]:
|
|
"""
|
|
Test if server accepts JWT with 'none' algorithm
|
|
|
|
Critical vulnerability: allows unsigned tokens to be accepted.
|
|
|
|
Returns:
|
|
dict[str, Any]: None algorithm test results
|
|
"""
|
|
try:
|
|
header, payload, signature = self.auth_token.split(".")
|
|
|
|
none_variants = AuthPayloads.get_jwt_none_variants()
|
|
|
|
for variant in none_variants:
|
|
malicious_header = self._base64url_encode(
|
|
json.dumps({
|
|
"alg": variant,
|
|
"typ": "JWT"
|
|
})
|
|
)
|
|
|
|
malicious_token = f"{malicious_header}.{payload}."
|
|
|
|
response = self.make_request(
|
|
"GET",
|
|
"/",
|
|
headers = {
|
|
"Authorization": f"Bearer {malicious_token}"
|
|
},
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return {
|
|
"vulnerable":
|
|
True,
|
|
"vulnerability_type":
|
|
"JWT None Algorithm",
|
|
"algorithm_variant":
|
|
variant,
|
|
"status_code":
|
|
response.status_code,
|
|
"recommendations": [
|
|
"Reject tokens with 'none' algorithm (all case variations)",
|
|
"Explicitly verify signature before accepting tokens",
|
|
"Use allowlist of accepted algorithms",
|
|
],
|
|
}
|
|
|
|
return {
|
|
"vulnerable": False,
|
|
"description": "None algorithm properly rejected",
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"vulnerable": False,
|
|
"error": str(e),
|
|
"description": "Error testing none algorithm",
|
|
}
|
|
|
|
def _test_signature_removal(self) -> dict[str, Any]:
|
|
"""
|
|
Test if server accepts JWT with signature removed
|
|
|
|
Returns:
|
|
dict[str, Any]: Signature removal test results
|
|
"""
|
|
try:
|
|
header, payload, signature = self.auth_token.split(".")
|
|
|
|
malicious_token = f"{header}.{payload}."
|
|
|
|
response = self.make_request(
|
|
"GET",
|
|
"/",
|
|
headers = {
|
|
"Authorization": f"Bearer {malicious_token}"
|
|
},
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return {
|
|
"vulnerable":
|
|
True,
|
|
"vulnerability_type":
|
|
"JWT Signature Not Verified",
|
|
"status_code":
|
|
response.status_code,
|
|
"recommendations": [
|
|
"Require valid signature on all JWT tokens",
|
|
"Reject tokens with missing or invalid signatures",
|
|
"Implement proper JWT validation library",
|
|
],
|
|
}
|
|
|
|
return {
|
|
"vulnerable": False,
|
|
"description": "Signature removal properly rejected",
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"vulnerable": False,
|
|
"error": str(e),
|
|
"description": "Error testing signature removal",
|
|
}
|
|
|
|
def _test_invalid_token_handling(self) -> dict[str, Any]:
|
|
"""
|
|
Test how server handles invalid/malformed tokens
|
|
|
|
Returns:
|
|
dict[str, Any]: Invalid token handling test results
|
|
"""
|
|
invalid_tokens = AuthPayloads.INVALID_TOKEN_FORMATS
|
|
|
|
accepted_invalid = []
|
|
|
|
for invalid_token in invalid_tokens:
|
|
try:
|
|
response = self.make_request(
|
|
"GET",
|
|
"/",
|
|
headers = {
|
|
"Authorization": f"Bearer {invalid_token}"
|
|
},
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
accepted_invalid.append(
|
|
{
|
|
"token": invalid_token[: 50],
|
|
"status_code": response.status_code,
|
|
}
|
|
)
|
|
|
|
except Exception:
|
|
continue
|
|
|
|
if accepted_invalid:
|
|
return {
|
|
"vulnerable": True,
|
|
"accepted_invalid_tokens": accepted_invalid,
|
|
"count": len(accepted_invalid),
|
|
}
|
|
|
|
return {
|
|
"vulnerable": False,
|
|
"description": "Invalid tokens properly rejected",
|
|
"tokens_tested": len(invalid_tokens),
|
|
}
|
|
|
|
def _base64url_decode(self, data: str) -> dict[str, Any]:
|
|
"""
|
|
Decode base64url-encoded JWT data
|
|
|
|
Args:
|
|
data: Base64url-encoded string
|
|
|
|
Returns:
|
|
dict[str, Any]: Decoded JSON data
|
|
"""
|
|
padding = 4 - (len(data) % 4)
|
|
if padding != 4:
|
|
data += "=" * padding
|
|
|
|
decoded = base64.urlsafe_b64decode(data)
|
|
return json.loads(decoded)
|
|
|
|
def _base64url_encode(self, data: str) -> str:
|
|
"""
|
|
Encode data to base64url format
|
|
|
|
Args:
|
|
data: String data to encode
|
|
|
|
Returns:
|
|
str: Base64url-encoded string
|
|
"""
|
|
encoded = base64.urlsafe_b64encode(data.encode()).decode()
|
|
return encoded.rstrip("=")
|
|
|
|
def _create_vulnerable_result(
|
|
self,
|
|
details: str,
|
|
evidence: dict[str,
|
|
Any],
|
|
severity: Severity = Severity.HIGH,
|
|
recommendations: list[str] | None = None,
|
|
) -> TestResultCreate:
|
|
"""
|
|
Create a vulnerable scan result
|
|
|
|
Args:
|
|
details: Vulnerability description
|
|
evidence: Evidence dictionary
|
|
severity: Vulnerability severity
|
|
recommendations: List of remediation recommendations
|
|
|
|
Returns:
|
|
TestResultCreate: Vulnerable result
|
|
"""
|
|
return TestResultCreate(
|
|
test_name = TestType.AUTH,
|
|
status = ScanStatus.VULNERABLE,
|
|
severity = severity,
|
|
details = details,
|
|
evidence_json = evidence,
|
|
recommendations_json = recommendations or [],
|
|
)
|