diff --git a/api-security-scanner/backend/config.py b/api-security-scanner/backend/config.py index b834ac09..98ab3205 100644 --- a/api-security-scanner/backend/config.py +++ b/api-security-scanner/backend/config.py @@ -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() diff --git a/api-security-scanner/backend/factory.py b/api-security-scanner/backend/factory.py index 25323df5..b72a1bd2 100644 --- a/api-security-scanner/backend/factory.py +++ b/api-security-scanner/backend/factory.py @@ -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) - diff --git a/api-security-scanner/backend/main.py b/api-security-scanner/backend/main.py index 70a778a5..65079c98 100644 --- a/api-security-scanner/backend/main.py +++ b/api-security-scanner/backend/main.py @@ -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, ) diff --git a/api-security-scanner/backend/routes/__init__.py b/api-security-scanner/backend/routes/__init__.py index 73f9851e..3e4a6304 100644 --- a/api-security-scanner/backend/routes/__init__.py +++ b/api-security-scanner/backend/routes/__init__.py @@ -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", diff --git a/api-security-scanner/backend/routes/auth.py b/api-security-scanner/backend/routes/auth.py index aa3806d1..10bc61b1 100644 --- a/api-security-scanner/backend/routes/auth.py +++ b/api-security-scanner/backend/routes/auth.py @@ -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) - diff --git a/api-security-scanner/backend/routes/scans.py b/api-security-scanner/backend/routes/scans.py index c7698db7..2bb4707b 100644 --- a/api-security-scanner/backend/routes/scans.py +++ b/api-security-scanner/backend/routes/scans.py @@ -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( diff --git a/api-security-scanner/backend/scanners/auth_scanner.py b/api-security-scanner/backend/scanners/auth_scanner.py index 575705ca..8aa3d1ad 100644 --- a/api-security-scanner/backend/scanners/auth_scanner.py +++ b/api-security-scanner/backend/scanners/auth_scanner.py @@ -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 [], ) diff --git a/api-security-scanner/backend/scanners/base_scanner.py b/api-security-scanner/backend/scanners/base_scanner.py index 9279b7bc..5c932ce7 100644 --- a/api-security-scanner/backend/scanners/base_scanner.py +++ b/api-security-scanner/backend/scanners/base_scanner.py @@ -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 diff --git a/api-security-scanner/backend/scanners/idor_scanner.py b/api-security-scanner/backend/scanners/idor_scanner.py index 2592db69..68987555 100644 --- a/api-security-scanner/backend/scanners/idor_scanner.py +++ b/api-security-scanner/backend/scanners/idor_scanner.py @@ -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 [], ) diff --git a/api-security-scanner/backend/scanners/payloads.py b/api-security-scanner/backend/scanners/payloads.py index d275505d..b8bcb0fd 100644 --- a/api-security-scanner/backend/scanners/payloads.py +++ b/api-security-scanner/backend/scanners/payloads.py @@ -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", } diff --git a/api-security-scanner/backend/scanners/rate_limit_scanner.py b/api-security-scanner/backend/scanners/rate_limit_scanner.py index 06ddd7ae..95767fd1 100644 --- a/api-security-scanner/backend/scanners/rate_limit_scanner.py +++ b/api-security-scanner/backend/scanners/rate_limit_scanner.py @@ -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, diff --git a/api-security-scanner/backend/scanners/sqli_scanner.py b/api-security-scanner/backend/scanners/sqli_scanner.py index ff085be4..e681b5ed 100644 --- a/api-security-scanner/backend/scanners/sqli_scanner.py +++ b/api-security-scanner/backend/scanners/sqli_scanner.py @@ -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 [], )