RomanNum3ral
/
Cybersecurity-Projects
mirror of https://github.com/CarterPerez-dev/Cybersecurity-Projects.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

yapf

This commit is contained in:
CarterPerez-dev 2025-11-10 13:53:22 -05:00
parent 7b0a43aaea
commit 9dd12f5a31
12 changed files with 283 additions and 195 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
api-security-scanner/backend/config.py
View File

@ -16,7 +16,9 @@ class Settings(BaseSettings):
"""
model_config = SettingsConfigDict(
env_file="../.env", env_file_encoding="utf-8", case_sensitive=True
env_file = "../.env",
env_file_encoding = "utf-8",
case_sensitive = True
)
# Application metadata
@ -84,7 +86,9 @@ class Settings(BaseSettings):
"""
Convert comma separated CORS origins string to list
"""
return [origin.strip() for origin in self.CORS_ORIGINS.split(",")]
return [
origin.strip() for origin in self.CORS_ORIGINS.split(",")
]
@lru_cache
@ -100,4 +104,5 @@ def get_settings() -> Settings:
"""
return Settings()
settings = get_settings()

31
api-security-scanner/backend/factory.py
View File

@ -7,7 +7,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from slowapi import (
Limiter,
Limiter,
_rate_limit_exceeded_handler,
)
from slowapi.errors import RateLimitExceeded
@ -22,26 +22,29 @@ def create_app() -> FastAPI:
"""
Application factory function
"""
Base.metadata.create_all(bind=engine)
Base.metadata.create_all(bind = engine)
app = FastAPI(
title=settings.APP_NAME,
version=settings.VERSION,
docs_url="/api/docs",
redoc_url="/api/redoc",
debug=settings.DEBUG,
title = settings.APP_NAME,
version = settings.VERSION,
docs_url = "/api/docs",
redoc_url = "/api/redoc",
debug = settings.DEBUG,
)
limiter = Limiter(key_func=get_remote_address)
limiter = Limiter(key_func = get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_exception_handler(
RateLimitExceeded,
_rate_limit_exceeded_handler
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins_list,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
allow_origins = settings.cors_origins_list,
allow_credentials = True,
allow_methods = ["*"],
allow_headers = ["*"],
)
_register_routes(app)
@ -49,7 +52,6 @@ def create_app() -> FastAPI:
return app
def _register_routes(app: FastAPI) -> None:
"""
Register all application routes
@ -74,4 +76,3 @@ def _register_routes(app: FastAPI) -> None:
app.include_router(auth_router)
app.include_router(scans_router)

6
api-security-scanner/backend/main.py
View File

@ -14,7 +14,7 @@ app = create_app()
if __name__ == "__main__":
uvicorn.run(
"main:app",
host=settings.BACKEND_HOST,
port=settings.BACKEND_PORT,
reload=settings.DEBUG,
host = settings.BACKEND_HOST,
port = settings.BACKEND_PORT,
reload = settings.DEBUG,
)

1
api-security-scanner/backend/routes/__init__.py
View File

@ -5,6 +5,7 @@ API route handlers
from .auth import router as auth_router
from .scans import router as scans_router
__all__ = [
"auth_router",
"scans_router",

19
api-security-scanner/backend/routes/auth.py
View File

@ -4,9 +4,9 @@ Authentication routes - registration and login
"""
from fastapi import (
APIRouter,
Depends,
Request,
APIRouter,
Depends,
Request,
status,
)
from slowapi import Limiter
@ -24,14 +24,14 @@ from ..schemas.user_schemas import (
from ..services.auth_service import AuthService
router = APIRouter(prefix="/auth", tags=["authentication"])
limiter = Limiter(key_func=get_remote_address)
router = APIRouter(prefix = "/auth", tags = ["authentication"])
limiter = Limiter(key_func = get_remote_address)
@router.post(
"/register",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
response_model = UserResponse,
status_code = status.HTTP_201_CREATED,
)
@limiter.limit(settings.API_RATE_LIMIT_REGISTER)
async def register(
@ -47,8 +47,8 @@ async def register(
@router.post(
"/login",
response_model=TokenResponse,
status_code=status.HTTP_200_OK,
response_model = TokenResponse,
status_code = status.HTTP_200_OK,
)
@limiter.limit(settings.API_RATE_LIMIT_LOGIN)
async def login(
@ -60,4 +60,3 @@ async def login(
Authenticate user and receive JWT token
"""
return AuthService.login_user(db, login_data)

33
api-security-scanner/backend/routes/scans.py
View File

@ -4,9 +4,9 @@ Scan routes - create, retrieve, and manage security scans
"""
from fastapi import (
APIRouter,
Depends,
Request,
APIRouter,
Depends,
Request,
status,
)
from slowapi import Limiter
@ -17,21 +17,21 @@ from ..config import settings
from ..core.database import get_db
from ..core.dependencies import get_current_user
from ..schemas.scan_schemas import (
ScanRequest,
ScanRequest,
ScanResponse,
)
from ..schemas.user_schemas import UserResponse
from ..services.scan_service import ScanService
router = APIRouter(prefix="/scans", tags=["scans"])
limiter = Limiter(key_func=get_remote_address)
router = APIRouter(prefix = "/scans", tags = ["scans"])
limiter = Limiter(key_func = get_remote_address)
@router.post(
"/",
response_model=ScanResponse,
status_code=status.HTTP_201_CREATED,
response_model = ScanResponse,
status_code = status.HTTP_201_CREATED,
)
@limiter.limit(settings.API_RATE_LIMIT_SCAN)
async def create_scan(
@ -48,8 +48,8 @@ async def create_scan(
@router.get(
"/",
response_model=list[ScanResponse],
status_code=status.HTTP_200_OK,
response_model = list[ScanResponse],
status_code = status.HTTP_200_OK,
)
@limiter.limit(settings.API_RATE_LIMIT_DEFAULT)
async def get_user_scans(
@ -62,13 +62,18 @@ async def get_user_scans(
"""
Get all scans for the authenticated user
"""
return ScanService.get_user_scans(db, current_user.id, skip, limit)
return ScanService.get_user_scans(
db,
current_user.id,
skip,
limit
)
@router.get(
"/{scan_id}",
response_model=ScanResponse,
status_code=status.HTTP_200_OK,
response_model = ScanResponse,
status_code = status.HTTP_200_OK,
)
@limiter.limit(settings.API_RATE_LIMIT_DEFAULT)
async def get_scan(
@ -85,7 +90,7 @@ async def get_scan(
@router.delete(
"/{scan_id}",
status_code=status.HTTP_204_NO_CONTENT,
status_code = status.HTTP_204_NO_CONTENT,
)
@limiter.limit(settings.API_RATE_LIMIT_DEFAULT)
async def delete_scan(

126
api-security-scanner/backend/scanners/auth_scanner.py
View File

@ -35,7 +35,6 @@ class AuthScanner(BaseScanner):
Maps to OWASP API Security Top 10 2023: API2:2023
"""
def scan(self) -> TestResultCreate:
"""
Execute authentication tests
@ -46,10 +45,10 @@ class AuthScanner(BaseScanner):
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=[
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",
@ -60,10 +59,11 @@ class AuthScanner(BaseScanner):
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(
details =
f"JWT vulnerability: {jwt_test['vulnerability_type']}",
evidence = jwt_test,
severity = Severity.CRITICAL,
recommendations = jwt_test.get(
"recommendations",
[
"Properly validate JWT signatures",
@ -77,10 +77,10 @@ class AuthScanner(BaseScanner):
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=[
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",
@ -88,15 +88,15 @@ class AuthScanner(BaseScanner):
)
return TestResultCreate(
test_name=TestType.AUTH,
status=ScanStatus.SAFE,
severity=Severity.INFO,
details="Authentication properly implemented",
evidence_json={
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=[
recommendations_json = [
"Authentication is properly configured",
"Consider implementing additional security measures (2FA, refresh tokens)",
],
@ -114,7 +114,8 @@ class AuthScanner(BaseScanner):
session_without_auth = self.session.__class__()
session_without_auth.headers.update(
{
"User-Agent": f"{settings.APP_NAME}/{settings.VERSION}",
"User-Agent":
f"{settings.APP_NAME}/{settings.VERSION}",
"Accept": "application/json",
}
)
@ -122,22 +123,29 @@ class AuthScanner(BaseScanner):
try:
response = session_without_auth.get(
self.target_url,
timeout=settings.SCANNER_CONNECTION_TIMEOUT,
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",
"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",
"vulnerable":
False,
"status_code":
response.status_code,
"description":
"Endpoint properly requires authentication",
}
return {
@ -150,7 +158,8 @@ class AuthScanner(BaseScanner):
return {
"vulnerable": False,
"error": str(e),
"description": "Error testing authentication requirement",
"description":
"Error testing authentication requirement",
}
def _test_jwt_vulnerabilities(self) -> dict[str, Any]:
@ -181,7 +190,9 @@ class AuthScanner(BaseScanner):
return {
"vulnerable": False,
"tests_performed": ["none_algorithm", "signature_removal"],
"tests_performed":
["none_algorithm",
"signature_removal"],
"description": "No JWT vulnerabilities detected",
}
@ -201,7 +212,10 @@ class AuthScanner(BaseScanner):
for variant in none_variants:
malicious_header = self._base64url_encode(
json.dumps({"alg": variant, "typ": "JWT"})
json.dumps({
"alg": variant,
"typ": "JWT"
})
)
malicious_token = f"{malicious_header}.{payload}."
@ -209,15 +223,21 @@ class AuthScanner(BaseScanner):
response = self.make_request(
"GET",
"/",
headers={"Authorization": f"Bearer {malicious_token}"},
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,
"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",
@ -252,14 +272,19 @@ class AuthScanner(BaseScanner):
response = self.make_request(
"GET",
"/",
headers={"Authorization": f"Bearer {malicious_token}"},
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,
"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",
@ -295,13 +320,15 @@ class AuthScanner(BaseScanner):
response = self.make_request(
"GET",
"/",
headers={"Authorization": f"Bearer {invalid_token}"},
headers = {
"Authorization": f"Bearer {invalid_token}"
},
)
if response.status_code == 200:
accepted_invalid.append(
{
"token": invalid_token[:50],
"token": invalid_token[: 50],
"status_code": response.status_code,
}
)
@ -355,7 +382,8 @@ class AuthScanner(BaseScanner):
def _create_vulnerable_result(
self,
details: str,
evidence: dict[str, Any],
evidence: dict[str,
Any],
severity: Severity = Severity.HIGH,
recommendations: list[str] | None = None,
) -> TestResultCreate:
@ -372,10 +400,10 @@ class AuthScanner(BaseScanner):
TestResultCreate: Vulnerable result
"""
return TestResultCreate(
test_name=TestType.AUTH,
status=ScanStatus.VULNERABLE,
severity=severity,
details=details,
evidence_json=evidence,
recommendations_json=recommendations or [],
test_name = TestType.AUTH,
status = ScanStatus.VULNERABLE,
severity = severity,
details = details,
evidence_json = evidence,
recommendations_json = recommendations or [],
)

9
api-security-scanner/backend/scanners/base_scanner.py
View File

@ -87,7 +87,8 @@ class BaseScanner(ABC):
jitter_ms = settings.DEFAULT_JITTER_MS
required_delay = 1.0 / (
self.max_requests / settings.SCANNER_RATE_LIMIT_WINDOW_SECONDS
self.max_requests /
settings.SCANNER_RATE_LIMIT_WINDOW_SECONDS
)
jitter = random.uniform(0, jitter_ms / 1000.0)
@ -138,7 +139,11 @@ class BaseScanner(ABC):
try:
start_time = time.time()
response = self.session.request(method, url, **kwargs)
setattr(response, "request_time", time.time() - start_time)
setattr(
response,
"request_time",
time.time() - start_time
)
self.request_count += 1

87
api-security-scanner/backend/scanners/idor_scanner.py
View File

@ -30,7 +30,6 @@ class IDORScanner(BaseScanner):
Maps to OWASP API Security Top 10 2023: API1:2023
"""
def scan(self) -> TestResultCreate:
"""
Execute IDOR/BOLA tests
@ -42,10 +41,11 @@ class IDORScanner(BaseScanner):
if id_enumeration_test["vulnerable"]:
return self._create_vulnerable_result(
details=f"IDOR vulnerability detected: {id_enumeration_test['vulnerability_type']}",
evidence=id_enumeration_test,
severity=Severity.HIGH,
recommendations=[
details =
f"IDOR vulnerability detected: {id_enumeration_test['vulnerability_type']}",
evidence = id_enumeration_test,
severity = Severity.HIGH,
recommendations = [
"Implement proper authorization checks for all object access",
"Verify user owns/has permission to access requested resource",
"Use UUIDs instead of sequential IDs (but still check authorization)",
@ -58,10 +58,11 @@ class IDORScanner(BaseScanner):
if predictable_id_test["vulnerable"]:
return self._create_vulnerable_result(
details="Predictable ID patterns detected enabling enumeration",
evidence=predictable_id_test,
severity=Severity.MEDIUM,
recommendations=[
details =
"Predictable ID patterns detected enabling enumeration",
evidence = predictable_id_test,
severity = Severity.MEDIUM,
recommendations = [
"Use non-sequential, non-predictable identifiers (UUIDs)",
"Implement rate limiting on ID-based endpoints",
"Add authorization checks regardless of ID format",
@ -69,15 +70,15 @@ class IDORScanner(BaseScanner):
)
return TestResultCreate(
test_name=TestType.IDOR,
status=ScanStatus.SAFE,
severity=Severity.INFO,
details="No IDOR/BOLA vulnerabilities detected",
evidence_json={
test_name = TestType.IDOR,
status = ScanStatus.SAFE,
severity = Severity.INFO,
details = "No IDOR/BOLA vulnerabilities detected",
evidence_json = {
"id_enumeration_test": id_enumeration_test,
"predictable_id_test": predictable_id_test,
},
recommendations_json=[
recommendations_json = [
"Authorization checks appear to be in place",
"Continue monitoring for authorization bypasses",
],
@ -101,7 +102,9 @@ class IDORScanner(BaseScanner):
"description": "No IDs found in endpoint responses",
}
numeric_test = self._test_numeric_id_manipulation(extracted_ids)
numeric_test = self._test_numeric_id_manipulation(
extracted_ids
)
if numeric_test["vulnerable"]:
return numeric_test
@ -133,23 +136,31 @@ class IDORScanner(BaseScanner):
response_text = response.text
uuid_pattern = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'
uuids = re.findall(uuid_pattern, response_text, re.IGNORECASE)
uuids = re.findall(
uuid_pattern,
response_text,
re.IGNORECASE
)
numeric_id_pattern = r'"id"\s*:\s*(\d+)'
numeric_ids = re.findall(numeric_id_pattern, response_text)
numeric_ids = re.findall(
numeric_id_pattern,
response_text
)
ids = []
ids.extend(uuids[:3])
ids.extend([int(nid) for nid in numeric_ids[:3]])
ids.extend(uuids[: 3])
ids.extend([int(nid) for nid in numeric_ids[: 3]])
return ids
except Exception:
return []
def _test_numeric_id_manipulation(
self, extracted_ids: list[Any]
) -> dict[str, Any]:
def _test_numeric_id_manipulation(self,
extracted_ids: list[Any]
) -> dict[str,
Any]:
"""
Test numeric ID manipulation for IDOR
@ -160,7 +171,8 @@ class IDORScanner(BaseScanner):
dict[str, Any]: Numeric ID manipulation test results
"""
numeric_ids = [
id_val for id_val in extracted_ids if isinstance(id_val, int)
id_val for id_val in extracted_ids
if isinstance(id_val, int)
]
if not numeric_ids:
@ -208,9 +220,10 @@ class IDORScanner(BaseScanner):
"numeric_ids_tested": len(test_ids),
}
def _test_string_id_manipulation(
self, extracted_ids: list[Any]
) -> dict[str, Any]:
def _test_string_id_manipulation(self,
extracted_ids: list[Any]
) -> dict[str,
Any]:
"""
Test string/UUID ID manipulation for IDOR
@ -221,7 +234,8 @@ class IDORScanner(BaseScanner):
dict[str, Any]: String ID manipulation test results
"""
string_ids = [
id_val for id_val in extracted_ids if isinstance(id_val, str)
id_val for id_val in extracted_ids
if isinstance(id_val, str)
]
if not string_ids:
@ -291,7 +305,7 @@ class IDORScanner(BaseScanner):
"vulnerable": True,
"pattern_type": "Sequential IDs",
"id_difference": diff1,
"example_ids": numeric_ids1[:3],
"example_ids": numeric_ids1[: 3],
}
return {
@ -309,7 +323,8 @@ class IDORScanner(BaseScanner):
def _create_vulnerable_result(
self,
details: str,
evidence: dict[str, Any],
evidence: dict[str,
Any],
severity: Severity = Severity.HIGH,
recommendations: list[str] | None = None,
) -> TestResultCreate:
@ -326,10 +341,10 @@ class IDORScanner(BaseScanner):
TestResultCreate: Vulnerable result
"""
return TestResultCreate(
test_name=TestType.IDOR,
status=ScanStatus.VULNERABLE,
severity=severity,
details=details,
evidence_json=evidence,
recommendations_json=recommendations or [],
test_name = TestType.IDOR,
status = ScanStatus.VULNERABLE,
severity = severity,
details = details,
evidence_json = evidence,
recommendations_json = recommendations or [],
)

9
api-security-scanner/backend/scanners/payloads.py
View File

@ -277,9 +277,12 @@ class RateLimitBypassPayloads:
"""
HEADER_PATTERNS = {
"limit": r"x-ratelimit-limit|x-rate-limit-limit|ratelimit-limit",
"remaining": r"x-ratelimit-remaining|x-rate-limit-remaining|ratelimit-remaining",
"reset": r"x-ratelimit-reset|x-rate-limit-reset|ratelimit-reset",
"limit":
r"x-ratelimit-limit|x-rate-limit-limit|ratelimit-limit",
"remaining":
r"x-ratelimit-remaining|x-rate-limit-remaining|ratelimit-remaining",
"reset":
r"x-ratelimit-reset|x-rate-limit-reset|ratelimit-reset",
"retry_after": r"retry-after",
}

3
api-security-scanner/backend/scanners/rate_limit_scanner.py
View File

@ -110,7 +110,8 @@ class RateLimitScanner(BaseScanner):
Returns:
dict[str, Any]: Rate limiting detection results
"""
rate_limit_patterns = RateLimitBypassPayloads.get_header_patterns()
rate_limit_patterns = RateLimitBypassPayloads.get_header_patterns(
)
results = {
"rate_limit_detected": False,

145
api-security-scanner/backend/scanners/sqli_scanner.py
View File

@ -31,7 +31,6 @@ class SQLiScanner(BaseScanner):
Uses payloads covering MySQL, PostgreSQL, MSSQL, Oracle
"""
def scan(self) -> TestResultCreate:
"""
Execute SQL injection tests
@ -42,10 +41,11 @@ class SQLiScanner(BaseScanner):
error_based_test = self._test_error_based_sqli()
if error_based_test["vulnerable"]:
return self._create_vulnerable_result(
details=f"Error-based SQL injection detected: {error_based_test['database_type']}",
evidence=error_based_test,
severity=Severity.CRITICAL,
recommendations=[
details =
f"Error-based SQL injection detected: {error_based_test['database_type']}",
evidence = error_based_test,
severity = Severity.CRITICAL,
recommendations = [
"Use parameterized queries (prepared statements)",
"Never concatenate user input into SQL queries",
"Implement input validation and sanitization",
@ -57,10 +57,10 @@ class SQLiScanner(BaseScanner):
boolean_based_test = self._test_boolean_based_sqli()
if boolean_based_test["vulnerable"]:
return self._create_vulnerable_result(
details="Boolean-based blind SQL injection detected",
evidence=boolean_based_test,
severity=Severity.CRITICAL,
recommendations=[
details = "Boolean-based blind SQL injection detected",
evidence = boolean_based_test,
severity = Severity.CRITICAL,
recommendations = [
"Use parameterized queries for all database operations",
"Implement proper input validation",
"Avoid exposing different responses for true/false conditions",
@ -70,10 +70,11 @@ class SQLiScanner(BaseScanner):
time_based_test = self._test_time_based_sqli()
if time_based_test["vulnerable"]:
return self._create_vulnerable_result(
details=f"Time-based blind SQL injection detected: {time_based_test['database_type']}",
evidence=time_based_test,
severity=Severity.CRITICAL,
recommendations=[
details =
f"Time-based blind SQL injection detected: {time_based_test['database_type']}",
evidence = time_based_test,
severity = Severity.CRITICAL,
recommendations = [
"Use parameterized queries exclusively",
"Implement strict input validation",
"Monitor for unusual response time patterns",
@ -81,16 +82,16 @@ class SQLiScanner(BaseScanner):
)
return TestResultCreate(
test_name=TestType.SQLI,
status=ScanStatus.SAFE,
severity=Severity.INFO,
details="No SQL injection vulnerabilities detected",
evidence_json={
test_name = TestType.SQLI,
status = ScanStatus.SAFE,
severity = Severity.INFO,
details = "No SQL injection vulnerabilities detected",
evidence_json = {
"error_based_test": error_based_test,
"boolean_based_test": boolean_based_test,
"time_based_test": time_based_test,
},
recommendations_json=[
recommendations_json = [
"Continue using parameterized queries",
"Regularly update security testing",
],
@ -111,9 +112,7 @@ class SQLiScanner(BaseScanner):
for payload in basic_payloads:
try:
response = self.make_request(
"GET", f"/?id={payload}"
)
response = self.make_request("GET", f"/?id={payload}")
response_text_lower = response.text.lower()
@ -126,7 +125,8 @@ class SQLiScanner(BaseScanner):
"payload": payload,
"status_code": response.status_code,
"error_signature": signature,
"response_excerpt": response.text[:500],
"response_excerpt":
response.text[: 500],
}
except Exception:
@ -160,8 +160,14 @@ class SQLiScanner(BaseScanner):
}
boolean_payloads = SQLiPayloads.BOOLEAN_BASED_BLIND
true_payloads = [p for p in boolean_payloads if "AND '1'='1" in p or "AND 1=1" in p]
false_payloads = [p for p in boolean_payloads if "AND '1'='2" in p or "AND 1=2" in p or "AND 1=0" in p]
true_payloads = [
p for p in boolean_payloads
if "AND '1'='1" in p or "AND 1=1" in p
]
false_payloads = [
p for p in boolean_payloads if "AND '1'='2" in p
or "AND 1=2" in p or "AND 1=0" in p
]
true_lengths = []
for payload in true_payloads:
@ -180,14 +186,18 @@ class SQLiScanner(BaseScanner):
if length_diff > 100 and avg_true != avg_false:
return {
"vulnerable": True,
"baseline_length": baseline_length,
"true_condition_avg_length": avg_true,
"false_condition_avg_length": avg_false,
"length_difference": length_diff,
"confidence": "HIGH"
if length_diff > 500
else "MEDIUM",
"vulnerable":
True,
"baseline_length":
baseline_length,
"true_condition_avg_length":
avg_true,
"false_condition_avg_length":
avg_false,
"length_difference":
length_diff,
"confidence":
"HIGH" if length_diff > 500 else "MEDIUM",
}
return {
@ -203,9 +213,9 @@ class SQLiScanner(BaseScanner):
"description": "Error testing boolean-based SQLi",
}
def _test_time_based_sqli(
self, delay_seconds: int = 5
) -> dict[str, Any]:
def _test_time_based_sqli(self,
delay_seconds: int = 5) -> dict[str,
Any]:
"""
Test for time based blind SQL injection
@ -227,9 +237,14 @@ class SQLiScanner(BaseScanner):
all_time_payloads = SQLiPayloads.TIME_BASED_BLIND
delay_payloads = {
"mysql": [p for p in all_time_payloads if "SLEEP" in p],
"postgres": [p for p in all_time_payloads if "pg_sleep" in p],
"mssql": [p for p in all_time_payloads if "WAITFOR" in p],
"mysql":
[p for p in all_time_payloads if "SLEEP" in p],
"postgres": [
p for p in all_time_payloads if "pg_sleep" in p
],
"mssql": [
p for p in all_time_payloads if "WAITFOR" in p
],
}
for db_type, payloads in delay_payloads.items():
@ -241,9 +256,13 @@ class SQLiScanner(BaseScanner):
response = self.make_request(
"GET",
f"/?id={payload}",
timeout=delay_seconds + 10,
timeout = delay_seconds + 10,
)
elapsed = getattr(
response,
"request_time",
0.0
)
elapsed = getattr(response, "request_time", 0.0)
delay_times.append(elapsed)
except Exception:
@ -255,22 +274,27 @@ class SQLiScanner(BaseScanner):
if avg_delay >= expected_delay_time - 1:
confidence = (
"HIGH"
if avg_delay >= expected_delay_time
"HIGH" if avg_delay >= expected_delay_time
else "MEDIUM"
)
return {
"vulnerable": True,
"database_type": db_type,
"payload": payload,
"baseline_time": f"{baseline_mean:.3f}s",
"response_time": f"{avg_delay:.3f}s",
"expected_delay": f"{expected_delay_time:.3f}s",
"confidence": confidence,
"individual_times": [
f"{t:.3f}s" for t in delay_times
],
"vulnerable":
True,
"database_type":
db_type,
"payload":
payload,
"baseline_time":
f"{baseline_mean:.3f}s",
"response_time":
f"{avg_delay:.3f}s",
"expected_delay":
f"{expected_delay_time:.3f}s",
"confidence":
confidence,
"individual_times":
[f"{t:.3f}s" for t in delay_times],
}
return {
@ -290,7 +314,8 @@ class SQLiScanner(BaseScanner):
def _create_vulnerable_result(
self,
details: str,
evidence: dict[str, Any],
evidence: dict[str,
Any],
severity: Severity = Severity.CRITICAL,
recommendations: list[str] | None = None,
) -> TestResultCreate:
@ -307,10 +332,10 @@ class SQLiScanner(BaseScanner):
TestResultCreate: Vulnerable result
"""
return TestResultCreate(
test_name=TestType.SQLI,
status=ScanStatus.VULNERABLE,
severity=severity,
details=details,
evidence_json=evidence,
recommendations_json=recommendations or [],
test_name = TestType.SQLI,
status = ScanStatus.VULNERABLE,
severity = severity,
details = details,
evidence_json = evidence,
recommendations_json = recommendations or [],
)