diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 41bef79d..fd4bcbfb 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -9,13 +9,16 @@ repos: files: ^PROJECTS/api-security-scanner/backend/ exclude: ^PROJECTS/api-security-scanner/backend/(\.venv|__pycache__|\.pytest_cache)/ - - id: ruff-format - name: ruff format (backend) - files: ^PROJECTS/api-security-scanner/backend/ - exclude: ^PROJECTS/api-security-scanner/backend/(\.venv|__pycache__|\.pytest_cache)/ - - repo: local hooks: + - id: yapf + name: yapf format (backend) + entry: bash -c 'cd PROJECTS/api-security-scanner/backend && yapf -i -r -vv models/ repositories/ schemas/ scanners/ core/ factory/' + language: system + types: [python] + files: ^PROJECTS/api-security-scanner/backend/ + pass_filenames: false + - id: mypy name: mypy type check (backend) entry: bash -c 'cd PROJECTS/api-security-scanner/backend && mypy .' diff --git a/PROJECTS/api-security-scanner/backend/core/database.py b/PROJECTS/api-security-scanner/backend/core/database.py index ab3e9f4c..d90feeae 100644 --- a/PROJECTS/api-security-scanner/backend/core/database.py +++ b/PROJECTS/api-security-scanner/backend/core/database.py @@ -13,12 +13,16 @@ from config import settings # Database engine engine = create_engine( settings.DATABASE_URL, - pool_pre_ping=True, - echo=settings.DEBUG, + pool_pre_ping = True, + echo = settings.DEBUG, ) # Session factory -SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) +SessionLocal = sessionmaker( + autocommit = False, + autoflush = False, + bind = engine +) # Base class Base = declarative_base() diff --git a/PROJECTS/api-security-scanner/backend/core/dependencies.py b/PROJECTS/api-security-scanner/backend/core/dependencies.py index d7d19bd6..fa592d61 100644 --- a/PROJECTS/api-security-scanner/backend/core/dependencies.py +++ b/PROJECTS/api-security-scanner/backend/core/dependencies.py @@ -34,25 +34,25 @@ async def get_current_user( if email is None: raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid authentication credentials", - headers={"WWW-Authenticate": "Bearer"}, + status_code = status.HTTP_401_UNAUTHORIZED, + detail = "Invalid authentication credentials", + headers = {"WWW-Authenticate": "Bearer"}, ) user = UserRepository.get_by_email(db, email) if not user: raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="User not found", - headers={"WWW-Authenticate": "Bearer"}, + status_code = status.HTTP_401_UNAUTHORIZED, + detail = "User not found", + headers = {"WWW-Authenticate": "Bearer"}, ) return UserResponse.model_validate(user) except ValueError: raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid authentication credentials", - headers={"WWW-Authenticate": "Bearer"}, + status_code = status.HTTP_401_UNAUTHORIZED, + detail = "Invalid authentication credentials", + headers = {"WWW-Authenticate": "Bearer"}, ) from None diff --git a/PROJECTS/api-security-scanner/backend/core/security.py b/PROJECTS/api-security-scanner/backend/core/security.py index 6ff3456f..96feed6b 100644 --- a/PROJECTS/api-security-scanner/backend/core/security.py +++ b/PROJECTS/api-security-scanner/backend/core/security.py @@ -22,7 +22,10 @@ def hash_password(password: str) -> str: return hashed.decode("utf-8") -def verify_password(plain_password: str, hashed_password: str) -> bool: +def verify_password( + plain_password: str, + hashed_password: str +) -> bool: """ Verify a plain text password against a hashed password """ @@ -31,7 +34,11 @@ def verify_password(plain_password: str, hashed_password: str) -> bool: return bcrypt.checkpw(password_bytes, hashed_bytes) -def create_access_token(data: dict[str, str], expires_delta: timedelta | None = None) -> str: +def create_access_token( + data: dict[str, + str], + expires_delta: timedelta | None = None +) -> str: """ Create a JWT access token """ @@ -40,10 +47,16 @@ def create_access_token(data: dict[str, str], expires_delta: timedelta | None = if expires_delta: expire = datetime.utcnow() + expires_delta else: - expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) + expire = datetime.utcnow() + timedelta( + minutes = settings.ACCESS_TOKEN_EXPIRE_MINUTES + ) to_encode.update({"exp": expire}) - encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) + encoded_jwt = jwt.encode( + to_encode, + settings.SECRET_KEY, + algorithm = settings.ALGORITHM + ) return encoded_jwt @@ -52,7 +65,11 @@ def decode_token(token: str) -> dict[str, str]: Decode and verify a JWT token """ try: - payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) + payload = jwt.decode( + token, + settings.SECRET_KEY, + algorithms = [settings.ALGORITHM] + ) return payload except JWTError as e: raise ValueError(f"Invalid token: {str(e)}") from e diff --git a/PROJECTS/api-security-scanner/backend/models/Base.py b/PROJECTS/api-security-scanner/backend/models/Base.py index b0f9797d..8e3a2a0b 100644 --- a/PROJECTS/api-security-scanner/backend/models/Base.py +++ b/PROJECTS/api-security-scanner/backend/models/Base.py @@ -24,12 +24,20 @@ class BaseModel(Base): __abstract__ = True - id = Column(Integer, primary_key=True, index=True, autoincrement=True) - created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(UTC)) + id = Column( + Integer, + primary_key = True, + index = True, + autoincrement = True + ) + created_at = Column( + DateTime(timezone = True), + default = lambda: datetime.now(UTC) + ) updated_at = Column( - DateTime(timezone=True), - default=lambda: datetime.now(UTC), - onupdate=lambda: datetime.now(UTC), + DateTime(timezone = True), + default = lambda: datetime.now(UTC), + onupdate = lambda: datetime.now(UTC), ) @declared_attr @@ -46,7 +54,11 @@ class BaseModel(Base): Returns: dict: Dictionary representation of the model """ - return {column.name: getattr(self, column.name) for column in self.__table__.columns} + return { + column.name: getattr(self, + column.name) + for column in self.__table__.columns + } def update(self, **kwargs: Any) -> None: """ diff --git a/PROJECTS/api-security-scanner/backend/models/Scan.py b/PROJECTS/api-security-scanner/backend/models/Scan.py index 82c2cd39..ec784252 100644 --- a/PROJECTS/api-security-scanner/backend/models/Scan.py +++ b/PROJECTS/api-security-scanner/backend/models/Scan.py @@ -29,25 +29,26 @@ class Scan(BaseModel): user_id = Column( Integer, - ForeignKey("users.id", ondelete="CASCADE"), - nullable=False, - index=True, + ForeignKey("users.id", + ondelete = "CASCADE"), + nullable = False, + index = True, ) target_url = Column( String(settings.URL_MAX_LENGTH), - nullable=False, + nullable = False, ) scan_date = Column( - DateTime(timezone=True), - default=lambda: datetime.now(UTC), - nullable=False, + DateTime(timezone = True), + default = lambda: datetime.now(UTC), + nullable = False, ) - user = relationship("User", backref="scans") + user = relationship("User", backref = "scans") test_results = relationship( "TestResult", - back_populates="scan", - cascade="all, delete-orphan", + back_populates = "scan", + cascade = "all, delete-orphan", ) def __repr__(self) -> str: @@ -64,7 +65,10 @@ class Scan(BaseModel): Returns: bool: True if any test result is vulnerable """ - return any(result.status == "vulnerable" for result in self.test_results) + return any( + result.status == "vulnerable" + for result in self.test_results + ) @property def vulnerability_count(self) -> int: @@ -74,4 +78,7 @@ class Scan(BaseModel): Returns: int: Number of vulnerable test results """ - return sum(1 for result in self.test_results if result.status == "vulnerable") + return sum( + 1 for result in self.test_results + if result.status == "vulnerable" + ) diff --git a/PROJECTS/api-security-scanner/backend/models/TestResult.py b/PROJECTS/api-security-scanner/backend/models/TestResult.py index 2508805a..04ac8d61 100644 --- a/PROJECTS/api-security-scanner/backend/models/TestResult.py +++ b/PROJECTS/api-security-scanner/backend/models/TestResult.py @@ -30,30 +30,35 @@ class TestResult(BaseModel): scan_id = Column( Integer, - ForeignKey("scans.id", ondelete="CASCADE"), - nullable=False, - index=True, + ForeignKey("scans.id", + ondelete = "CASCADE"), + nullable = False, + index = True, ) test_name = Column( Enum(TestType), - nullable=False, - index=True, + nullable = False, + index = True, ) status = Column( Enum(ScanStatus), - nullable=False, - index=True, + nullable = False, + index = True, ) severity = Column( Enum(Severity), - nullable=False, - index=True, + nullable = False, + index = True, + ) + details = Column(Text, nullable = False) + evidence_json = Column(JSON, nullable = False, default = dict) + recommendations_json = Column( + JSON, + nullable = False, + default = list ) - details = Column(Text, nullable=False) - evidence_json = Column(JSON, nullable=False, default=dict) - recommendations_json = Column(JSON, nullable=False, default=list) - scan = relationship("Scan", back_populates="test_results") + scan = relationship("Scan", back_populates = "test_results") def __repr__(self) -> str: """ diff --git a/PROJECTS/api-security-scanner/backend/models/User.py b/PROJECTS/api-security-scanner/backend/models/User.py index 2935375f..54445262 100644 --- a/PROJECTS/api-security-scanner/backend/models/User.py +++ b/PROJECTS/api-security-scanner/backend/models/User.py @@ -22,12 +22,12 @@ class User(BaseModel): email = Column( String(settings.EMAIL_MAX_LENGTH), - unique=True, - nullable=False, - index=True, + unique = True, + nullable = False, + index = True, ) - hashed_password = Column(String, nullable=False) - is_active = Column(Boolean, default=True, nullable=False) + hashed_password = Column(String, nullable = False) + is_active = Column(Boolean, default = True, nullable = False) def __repr__(self) -> str: """ diff --git a/PROJECTS/api-security-scanner/backend/repositories/scan_repository.py b/PROJECTS/api-security-scanner/backend/repositories/scan_repository.py index 19465099..83376f0e 100644 --- a/PROJECTS/api-security-scanner/backend/repositories/scan_repository.py +++ b/PROJECTS/api-security-scanner/backend/repositories/scan_repository.py @@ -19,9 +19,13 @@ class ScanRepository: """ Repository for Scan database operations """ - @staticmethod - def create_scan(db: Session, user_id: int, target_url: str, commit: bool = True) -> Scan: + def create_scan( + db: Session, + user_id: int, + target_url: str, + commit: bool = True + ) -> Scan: """ Create a new scan @@ -35,9 +39,9 @@ class ScanRepository: Scan: Created scan instance """ scan = Scan( - user_id=user_id, - target_url=target_url, - scan_date=datetime.now(UTC), + user_id = user_id, + target_url = target_url, + scan_date = datetime.now(UTC), ) db.add(scan) if commit: @@ -58,15 +62,17 @@ class ScanRepository: Scan | None: Scan instance or None if not found """ return ( - db.query(Scan) - .options(joinedload(Scan.test_results)) - .filter(Scan.id == scan_id) - .first() + db.query(Scan).options( + joinedload(Scan.test_results) + ).filter(Scan.id == scan_id).first() ) @staticmethod def get_by_user( - db: Session, user_id: int, skip: int = 0, limit: int | None = None + db: Session, + user_id: int, + skip: int = 0, + limit: int | None = None ) -> list[Scan]: """ Get all scans for a user with pagination. @@ -84,17 +90,16 @@ class ScanRepository: limit = settings.DEFAULT_PAGINATION_LIMIT return ( - db.query(Scan) - .options(joinedload(Scan.test_results)) - .filter(Scan.user_id == user_id) - .order_by(Scan.scan_date.desc()) - .offset(skip) - .limit(limit) - .all() + db.query(Scan).options( + joinedload(Scan.test_results) + ).filter(Scan.user_id == user_id).order_by( + Scan.scan_date.desc() + ).offset(skip).limit(limit).all() ) @staticmethod - def get_recent(db: Session, limit: int | None = None) -> list[Scan]: + def get_recent(db: Session, + limit: int | None = None) -> list[Scan]: """ Get most recent scans across all users. @@ -109,15 +114,17 @@ class ScanRepository: limit = settings.DEFAULT_PAGINATION_LIMIT return ( - db.query(Scan) - .options(joinedload(Scan.test_results)) - .order_by(Scan.scan_date.desc()) - .limit(limit) - .all() + db.query(Scan).options( + joinedload(Scan.test_results) + ).order_by(Scan.scan_date.desc()).limit(limit).all() ) @staticmethod - def delete(db: Session, scan_id: int, commit: bool = True) -> bool: + def delete( + db: Session, + scan_id: int, + commit: bool = True + ) -> bool: """ Delete a scan (cascades to test results). diff --git a/PROJECTS/api-security-scanner/backend/repositories/test_result_repository.py b/PROJECTS/api-security-scanner/backend/repositories/test_result_repository.py index a15e4082..03f411d6 100644 --- a/PROJECTS/api-security-scanner/backend/repositories/test_result_repository.py +++ b/PROJECTS/api-security-scanner/backend/repositories/test_result_repository.py @@ -21,7 +21,6 @@ class TestResultRepository: """ Repository for TestResult database operations """ - @staticmethod def create_test_result( db: Session, @@ -31,7 +30,8 @@ class TestResultRepository: status: ScanStatus, severity: Severity, details: str, - evidence_json: dict[str, Any], + evidence_json: dict[str, + Any], recommendations_json: list[str], commit: bool = True, ) -> TestResult: @@ -53,13 +53,13 @@ class TestResultRepository: TestResult: Created test result instance """ test_result = TestResult( - scan_id=scan_id, - test_name=test_name, - status=status, - severity=severity, - details=details, - evidence_json=evidence_json, - recommendations_json=recommendations_json, + scan_id = scan_id, + test_name = test_name, + status = status, + severity = severity, + details = details, + evidence_json = evidence_json, + recommendations_json = recommendations_json, ) db.add(test_result) if commit: @@ -69,7 +69,9 @@ class TestResultRepository: @staticmethod def bulk_create( - db: Session, test_results: list[TestResult], commit: bool = True + db: Session, + test_results: list[TestResult], + commit: bool = True ) -> list[TestResult]: """ Create multiple test results in bulk @@ -102,14 +104,15 @@ class TestResultRepository: list[TestResult]: List of test results for the scan """ return ( - db.query(TestResult) - .filter(TestResult.scan_id == scan_id) - .order_by(TestResult.created_at.asc()) - .all() + db.query(TestResult).filter( + TestResult.scan_id == scan_id + ).order_by(TestResult.created_at.asc()).all() ) @staticmethod - def get_by_status(db: Session, scan_id: int, status: ScanStatus) -> list[TestResult]: + def get_by_status(db: Session, + scan_id: int, + status: ScanStatus) -> list[TestResult]: """ Get test results by status for a scan @@ -122,13 +125,15 @@ class TestResultRepository: list[TestResult]: Filtered test results """ return ( - db.query(TestResult) - .filter(TestResult.scan_id == scan_id, TestResult.status == status) - .all() + db.query(TestResult).filter( + TestResult.scan_id == scan_id, + TestResult.status == status + ).all() ) @staticmethod - def get_vulnerabilities(db: Session, scan_id: int) -> list[TestResult]: + def get_vulnerabilities(db: Session, + scan_id: int) -> list[TestResult]: """ Get only vulnerable test results for a scan @@ -139,10 +144,18 @@ class TestResultRepository: Returns: list[TestResult]: Vulnerable test results only """ - return TestResultRepository.get_by_status(db, scan_id, ScanStatus.VULNERABLE) + return TestResultRepository.get_by_status( + db, + scan_id, + ScanStatus.VULNERABLE + ) @staticmethod - def delete_by_scan(db: Session, scan_id: int, commit: bool = True) -> int: + def delete_by_scan( + db: Session, + scan_id: int, + commit: bool = True + ) -> int: """ Delete all test results for a scan @@ -154,7 +167,9 @@ class TestResultRepository: Returns: int: Number of test results deleted """ - count = db.query(TestResult).filter(TestResult.scan_id == scan_id).delete() + count = db.query(TestResult).filter( + TestResult.scan_id == scan_id + ).delete() if commit: db.commit() return count diff --git a/PROJECTS/api-security-scanner/backend/repositories/user_repository.py b/PROJECTS/api-security-scanner/backend/repositories/user_repository.py index 07663d5a..78896375 100644 --- a/PROJECTS/api-security-scanner/backend/repositories/user_repository.py +++ b/PROJECTS/api-security-scanner/backend/repositories/user_repository.py @@ -15,7 +15,6 @@ class UserRepository: """ Repository for User database operations """ - @staticmethod def get_by_id(db: Session, user_id: int) -> User | None: """ @@ -46,7 +45,10 @@ class UserRepository: @staticmethod def create_user( - db: Session, email: str, hashed_password: str, commit: bool = True + db: Session, + email: str, + hashed_password: str, + commit: bool = True ) -> User: """ Create a new user @@ -60,7 +62,7 @@ class UserRepository: Returns: User: Created user instance """ - user = User(email=email, hashed_password=hashed_password) + user = User(email = email, hashed_password = hashed_password) db.add(user) if commit: db.commit() @@ -68,7 +70,11 @@ class UserRepository: return user @staticmethod - def get_all_active(db: Session, skip: int = 0, limit: int | None = None) -> list[User]: + def get_all_active( + db: Session, + skip: int = 0, + limit: int | None = None + ) -> list[User]: """ Get all active users with pagination @@ -83,11 +89,15 @@ class UserRepository: if limit is None: limit = settings.DEFAULT_PAGINATION_LIMIT - return db.query(User).filter(User.is_active).offset(skip).limit(limit).all() + return db.query(User).filter(User.is_active + ).offset(skip).limit(limit).all() @staticmethod def update_active_status( - db: Session, user_id: int, is_active: bool, commit: bool = True + db: Session, + user_id: int, + is_active: bool, + commit: bool = True ) -> User | None: """ Update user active status @@ -110,7 +120,11 @@ class UserRepository: return user @staticmethod - def delete(db: Session, user_id: int, commit: bool = True) -> bool: + def delete( + db: Session, + user_id: int, + commit: bool = True + ) -> bool: """ Delete a user diff --git a/PROJECTS/api-security-scanner/backend/scanners/auth_scanner.py b/PROJECTS/api-security-scanner/backend/scanners/auth_scanner.py index ff57ce8d..14604a87 100644 --- a/PROJECTS/api-security-scanner/backend/scanners/auth_scanner.py +++ b/PROJECTS/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/PROJECTS/api-security-scanner/backend/scanners/base_scanner.py b/PROJECTS/api-security-scanner/backend/scanners/base_scanner.py index 7a43ae2c..28675978 100644 --- a/PROJECTS/api-security-scanner/backend/scanners/base_scanner.py +++ b/PROJECTS/api-security-scanner/backend/scanners/base_scanner.py @@ -25,7 +25,6 @@ class BaseScanner(ABC): Provides common HTTP functionality, request spacing, retry logic, and evidence collection. Specific scanners inherit and implement scan(). """ - def __init__( self, target_url: str, @@ -58,17 +57,23 @@ class BaseScanner(ABC): session.headers.update( { - "User-Agent": f"{settings.APP_NAME}/{settings.VERSION}", + "User-Agent": + f"{settings.APP_NAME}/{settings.VERSION}", "Accept": "application/json", } ) if self.auth_token: - session.headers.update({"Authorization": f"Bearer {self.auth_token}"}) + session.headers.update( + {"Authorization": f"Bearer {self.auth_token}"} + ) return session - def _wait_before_request(self, jitter_ms: int | None = None) -> None: + def _wait_before_request( + self, + jitter_ms: int | None = None + ) -> None: """ Implement request spacing to avoid overwhelming target @@ -81,7 +86,10 @@ class BaseScanner(ABC): if jitter_ms is None: jitter_ms = settings.DEFAULT_JITTER_MS - required_delay = 1.0 / (self.max_requests / settings.SCANNER_RATE_LIMIT_WINDOW_SECONDS) + required_delay = 1.0 / ( + self.max_requests / + settings.SCANNER_RATE_LIMIT_WINDOW_SECONDS + ) jitter = random.uniform(0, jitter_ms / 1000.0) elapsed = time.time() - self.last_request_time @@ -122,24 +130,31 @@ class BaseScanner(ABC): retry_count = 0 backoff_factor = 2.0 - kwargs.setdefault("timeout", settings.SCANNER_CONNECTION_TIMEOUT) + kwargs.setdefault( + "timeout", + settings.SCANNER_CONNECTION_TIMEOUT + ) while retry_count <= settings.DEFAULT_RETRY_COUNT: 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 if response.status_code == 429: retry_after = response.headers.get( - "Retry-After", str(settings.DEFAULT_RETRY_WAIT_SECONDS) + "Retry-After", + str(settings.DEFAULT_RETRY_WAIT_SECONDS) ) wait_time = ( - int(retry_after) - if retry_after.isdigit() - else settings.DEFAULT_RETRY_WAIT_SECONDS + int(retry_after) if retry_after.isdigit() else + settings.DEFAULT_RETRY_WAIT_SECONDS ) time.sleep(wait_time) retry_count += 1 @@ -164,8 +179,11 @@ class BaseScanner(ABC): return response def get_baseline_timing( - self, endpoint: str, samples: int | None = None - ) -> tuple[float, float]: + self, + endpoint: str, + samples: int | None = None + ) -> tuple[float, + float]: """ Establish baseline response time for an endpoint @@ -196,7 +214,8 @@ class BaseScanner(ABC): response: requests.Response, payload: Any | None = None, **additional_data: Any, - ) -> dict[str, Any]: + ) -> dict[str, + Any]: """ Collect evidence from test execution with sensitive data redaction @@ -209,10 +228,17 @@ class BaseScanner(ABC): dict[str, Any]: Evidence dictionary """ evidence = { - "status_code": response.status_code, - "response_time_ms": round(getattr(response, "request_time", 0.0) * 1000, 2), - "response_length": len(response.text), - "headers": self._redact_sensitive_headers(dict(response.headers)), + "status_code": + response.status_code, + "response_time_ms": + round(getattr(response, + "request_time", + 0.0) * 1000, + 2), + "response_length": + len(response.text), + "headers": + self._redact_sensitive_headers(dict(response.headers)), } if payload is not None: @@ -222,7 +248,10 @@ class BaseScanner(ABC): return evidence - def _redact_sensitive_headers(self, headers: dict[str, str]) -> dict[str, str]: + def _redact_sensitive_headers(self, + headers: dict[str, + str]) -> dict[str, + str]: """ Redact sensitive header values for evidence collection diff --git a/PROJECTS/api-security-scanner/backend/scanners/idor_scanner.py b/PROJECTS/api-security-scanner/backend/scanners/idor_scanner.py index 9f363f36..705e4aa2 100644 --- a/PROJECTS/api-security-scanner/backend/scanners/idor_scanner.py +++ b/PROJECTS/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,21 +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 @@ -157,7 +170,10 @@ class IDORScanner(BaseScanner): Returns: dict[str, Any]: Numeric ID manipulation test results """ - numeric_ids = [id_val for id_val in extracted_ids if isinstance(id_val, int)] + numeric_ids = [ + id_val for id_val in extracted_ids + if isinstance(id_val, int) + ] if not numeric_ids: return { @@ -204,7 +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 @@ -214,7 +233,10 @@ class IDORScanner(BaseScanner): Returns: dict[str, Any]: String ID manipulation test results """ - string_ids = [id_val for id_val in extracted_ids if isinstance(id_val, str)] + string_ids = [ + id_val for id_val in extracted_ids + if isinstance(id_val, str) + ] if not string_ids: return { @@ -283,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 { @@ -301,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: @@ -318,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/PROJECTS/api-security-scanner/backend/scanners/payloads.py b/PROJECTS/api-security-scanner/backend/scanners/payloads.py index 1c884849..05cead97 100644 --- a/PROJECTS/api-security-scanner/backend/scanners/payloads.py +++ b/PROJECTS/api-security-scanner/backend/scanners/payloads.py @@ -131,13 +131,10 @@ class SQLiPayloads: list[str]: All SQLi test payloads """ return ( - cls.BASIC_AUTHENTICATION_BYPASS - + cls.UNION_BASED - + cls.TIME_BASED_BLIND - + cls.BOOLEAN_BASED_BLIND - + cls.ERROR_BASED - + cls.STACKED_QUERIES - + cls.COMMENT_VARIATIONS + cls.BASIC_AUTHENTICATION_BYPASS + cls.UNION_BASED + + cls.TIME_BASED_BLIND + cls.BOOLEAN_BASED_BLIND + + cls.ERROR_BASED + cls.STACKED_QUERIES + + cls.COMMENT_VARIATIONS ) @classmethod @@ -280,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", } @@ -300,14 +300,30 @@ class RateLimitBypassPayloads: ] HEADER_SPOOFING = [ - {"X-Forwarded-For": "127.0.0.1"}, - {"X-Forwarded-For": "8.8.8.8"}, - {"X-Real-IP": "127.0.0.1"}, - {"X-Originating-IP": "127.0.0.1"}, - {"X-Remote-IP": "127.0.0.1"}, - {"X-Client-IP": "127.0.0.1"}, - {"CF-Connecting-IP": "127.0.0.1"}, - {"True-Client-IP": "127.0.0.1"}, + { + "X-Forwarded-For": "127.0.0.1" + }, + { + "X-Forwarded-For": "8.8.8.8" + }, + { + "X-Real-IP": "127.0.0.1" + }, + { + "X-Originating-IP": "127.0.0.1" + }, + { + "X-Remote-IP": "127.0.0.1" + }, + { + "X-Client-IP": "127.0.0.1" + }, + { + "CF-Connecting-IP": "127.0.0.1" + }, + { + "True-Client-IP": "127.0.0.1" + }, ] USER_AGENT_ROTATION = [ @@ -433,14 +449,10 @@ class XSSPayloads: list[str]: All XSS test payloads """ return ( - cls.BASIC_XSS - + cls.EVENT_HANDLER_XSS - + cls.SVG_XSS - + cls.IFRAME_XSS - + cls.ENCODED_XSS - + cls.ATTRIBUTE_BREAKING - + cls.FILTER_BYPASS - + cls.POLYGLOT_XSS + cls.BASIC_XSS + cls.EVENT_HANDLER_XSS + cls.SVG_XSS + + cls.IFRAME_XSS + cls.ENCODED_XSS + + cls.ATTRIBUTE_BREAKING + cls.FILTER_BYPASS + + cls.POLYGLOT_XSS ) @classmethod diff --git a/PROJECTS/api-security-scanner/backend/scanners/rate_limit_scanner.py b/PROJECTS/api-security-scanner/backend/scanners/rate_limit_scanner.py index fd09356f..168d1b05 100644 --- a/PROJECTS/api-security-scanner/backend/scanners/rate_limit_scanner.py +++ b/PROJECTS/api-security-scanner/backend/scanners/rate_limit_scanner.py @@ -26,7 +26,6 @@ class RateLimitScanner(BaseScanner): """ Rate limiting and bypass vulnerabilities tests """ - def scan(self) -> TestResultCreate: """ Execute rate limiting tests @@ -38,9 +37,10 @@ class RateLimitScanner(BaseScanner): if not rate_limit_info["rate_limit_detected"]: return self._create_vulnerable_result( - details="No rate limiting detected on target endpoint", - evidence=rate_limit_info, - recommendations=[ + details = + "No rate limiting detected on target endpoint", + evidence = rate_limit_info, + recommendations = [ "Implement rate limiting to prevent abuse and DoS attacks", "Use standard rate limit headers (X-RateLimit-Limit, X-RateLimit-Remaining)", "Return 429 Too Many Requests when limits are exceeded", @@ -50,10 +50,11 @@ class RateLimitScanner(BaseScanner): if rate_limit_info["enforcement_status"] == "HEADERS_ONLY": return self._create_vulnerable_result( - details="Rate limit headers present but not enforced", - evidence=rate_limit_info, - severity=Severity.MEDIUM, - recommendations=[ + details = + "Rate limit headers present but not enforced", + evidence = rate_limit_info, + severity = Severity.MEDIUM, + recommendations = [ "Enforce rate limits with 429 responses when thresholds are exceeded", "Rate limit headers without enforcement provide false security", ], @@ -63,13 +64,14 @@ class RateLimitScanner(BaseScanner): if bypass_results["bypass_successful"]: return self._create_vulnerable_result( - details=f"Rate limiting bypassed using: {bypass_results['bypass_method']}", - evidence={ + details = + f"Rate limiting bypassed using: {bypass_results['bypass_method']}", + evidence = { "rate_limit_info": rate_limit_info, "bypass_details": bypass_results, }, - severity=Severity.HIGH, - recommendations=[ + severity = Severity.HIGH, + recommendations = [ f"Fix bypass vulnerability: {bypass_results['bypass_method']}", "Do not trust client-provided IP headers (X-Forwarded-For, X-Real-IP)", "Implement rate limiting at multiple layers (IP, user, API key)", @@ -78,21 +80,25 @@ class RateLimitScanner(BaseScanner): ) return TestResultCreate( - test_name=TestType.RATE_LIMIT, - status=ScanStatus.SAFE, - severity=Severity.INFO, - details="Rate limiting properly implemented and enforced", - evidence_json={ + test_name = TestType.RATE_LIMIT, + status = ScanStatus.SAFE, + severity = Severity.INFO, + details = + "Rate limiting properly implemented and enforced", + evidence_json = { "rate_limit_info": rate_limit_info, "bypass_attempts": bypass_results, }, - recommendations_json=[ + recommendations_json = [ "Rate limiting is properly configured", "Continue monitoring for new bypass techniques", ], ) - def _detect_rate_limiting(self, test_request_count: int = 20) -> dict[str, Any]: + def _detect_rate_limiting(self, + test_request_count: int = 20 + ) -> dict[str, + Any]: """ Detect rate limiting by analyzing headers and response patterns @@ -104,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, @@ -120,23 +127,35 @@ class RateLimitScanner(BaseScanner): try: response = self.make_request("GET", "/") - headers_lower = {k.lower(): v for k, v in response.headers.items()} + headers_lower = { + k.lower(): v + for k, v in response.headers.items() + } for header_type, pattern in rate_limit_patterns.items(): for header_name, header_value in headers_lower.items(): - if re.search(pattern, header_name, re.IGNORECASE): - results["rate_limit_headers"][header_type] = { - "header_name": header_name, - "value": header_value, - } + if re.search(pattern, + header_name, + re.IGNORECASE): + results["rate_limit_headers"][ + header_type] = { + "header_name": header_name, + "value": header_value, + } results["rate_limit_detected"] = True results["request_results"].append( { - "attempt": attempt, - "status_code": response.status_code, - "response_time_ms": round( - getattr(response, "request_time", 0.0) * 1000, 2 + "attempt": + attempt, + "status_code": + response.status_code, + "response_time_ms": + round( + getattr(response, + "request_time", + 0.0) * 1000, + 2 ), } ) @@ -154,15 +173,22 @@ class RateLimitScanner(BaseScanner): time.sleep(0.1) except Exception as e: - results["request_results"].append({"attempt": attempt, "error": str(e)}) + results["request_results"].append( + { + "attempt": attempt, + "error": str(e) + } + ) break if results["rate_limit_detected"]: if "limit" in results["rate_limit_headers"]: - results["limit_threshold"] = results["rate_limit_headers"]["limit"]["value"] + results["limit_threshold"] = results[ + "rate_limit_headers"]["limit"]["value"] if "reset" in results["rate_limit_headers"]: - results["reset_window"] = results["rate_limit_headers"]["reset"]["value"] + results["reset_window"] = results["rate_limit_headers" + ]["reset"]["value"] if not results["enforcement_status"]: results["enforcement_status"] = "HEADERS_ONLY" @@ -209,7 +235,9 @@ class RateLimitScanner(BaseScanner): return results - def _test_ip_header_bypass(self, test_count: int = 15) -> dict[str, Any]: + def _test_ip_header_bypass(self, + test_count: int = 15) -> dict[str, + Any]: """ Test if rate limiting can be bypassed with IP spoofing headers @@ -233,7 +261,11 @@ class RateLimitScanner(BaseScanner): test_headers = {header_name: fake_ip} try: - response = self.make_request("GET", "/", headers=test_headers) + response = self.make_request( + "GET", + "/", + headers = test_headers + ) if response.status_code != 429: success_count += 1 @@ -253,7 +285,8 @@ class RateLimitScanner(BaseScanner): return { "bypass_successful": False, - "headers_tested": [list(h.keys())[0] for h in bypass_headers], + "headers_tested": + [list(h.keys())[0] for h in bypass_headers], } def _test_endpoint_variation_bypass(self) -> dict[str, Any]: @@ -295,7 +328,8 @@ class RateLimitScanner(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: @@ -312,10 +346,10 @@ class RateLimitScanner(BaseScanner): TestResultCreate: Vulnerable result """ return TestResultCreate( - test_name=TestType.RATE_LIMIT, - status=ScanStatus.VULNERABLE, - severity=severity, - details=details, - evidence_json=evidence, - recommendations_json=recommendations or [], + test_name = TestType.RATE_LIMIT, + status = ScanStatus.VULNERABLE, + severity = severity, + details = details, + evidence_json = evidence, + recommendations_json = recommendations or [], ) diff --git a/PROJECTS/api-security-scanner/backend/scanners/sqli_scanner.py b/PROJECTS/api-security-scanner/backend/scanners/sqli_scanner.py index 6a9d73d6..cf04feea 100644 --- a/PROJECTS/api-security-scanner/backend/scanners/sqli_scanner.py +++ b/PROJECTS/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", ], @@ -124,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: @@ -159,12 +161,12 @@ 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 + 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 + 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 = [] @@ -184,12 +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 { @@ -205,7 +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: @@ -257,14 +276,22 @@ class SQLiScanner(BaseScanner): confidence = "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 { @@ -284,7 +311,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: @@ -301,10 +329,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 [], ) diff --git a/PROJECTS/api-security-scanner/backend/schemas/scan_schemas.py b/PROJECTS/api-security-scanner/backend/schemas/scan_schemas.py index f1a21aea..d8675457 100644 --- a/PROJECTS/api-security-scanner/backend/schemas/scan_schemas.py +++ b/PROJECTS/api-security-scanner/backend/schemas/scan_schemas.py @@ -23,13 +23,13 @@ class ScanRequest(BaseModel): Schema for creating a new security scan """ - target_url: HttpUrl = Field(max_length=settings.URL_MAX_LENGTH) + target_url: HttpUrl = Field(max_length = settings.URL_MAX_LENGTH) auth_token: str | None = None - tests_to_run: list[TestType] = Field(min_length=1) + tests_to_run: list[TestType] = Field(min_length = 1) max_requests: int = Field( - default=settings.DEFAULT_MAX_REQUESTS, - ge=1, - le=settings.SCANNER_MAX_CONCURRENT_REQUESTS, + default = settings.DEFAULT_MAX_REQUESTS, + ge = 1, + le = settings.SCANNER_MAX_CONCURRENT_REQUESTS, ) @@ -38,7 +38,7 @@ class ScanResponse(BaseModel): Schema for scan data in API responses """ - model_config = ConfigDict(from_attributes=True) + model_config = ConfigDict(from_attributes = True) id: int user_id: int @@ -59,4 +59,6 @@ class ScanResponse(BaseModel): """ Number of vulnerabilities found """ - return sum(1 for r in self.test_results if r.status == "vulnerable") + return sum( + 1 for r in self.test_results if r.status == "vulnerable" + ) diff --git a/PROJECTS/api-security-scanner/backend/schemas/test_result_schemas.py b/PROJECTS/api-security-scanner/backend/schemas/test_result_schemas.py index f24e037d..fd549b87 100644 --- a/PROJECTS/api-security-scanner/backend/schemas/test_result_schemas.py +++ b/PROJECTS/api-security-scanner/backend/schemas/test_result_schemas.py @@ -27,8 +27,8 @@ class TestResultCreate(BaseModel): status: ScanStatus severity: Severity details: str - evidence_json: dict[str, Any] = Field(default_factory=dict) - recommendations_json: list[str] = Field(default_factory=list) + evidence_json: dict[str, Any] = Field(default_factory = dict) + recommendations_json: list[str] = Field(default_factory = list) class TestResultResponse(BaseModel): @@ -36,7 +36,7 @@ class TestResultResponse(BaseModel): Schema for individual test result in API responses """ - model_config = ConfigDict(from_attributes=True) + model_config = ConfigDict(from_attributes = True) id: int scan_id: int diff --git a/PROJECTS/api-security-scanner/backend/schemas/user_schemas.py b/PROJECTS/api-security-scanner/backend/schemas/user_schemas.py index 25decfd0..92c9907b 100644 --- a/PROJECTS/api-security-scanner/backend/schemas/user_schemas.py +++ b/PROJECTS/api-security-scanner/backend/schemas/user_schemas.py @@ -19,8 +19,8 @@ class UserCreate(BaseModel): email: EmailStr password: str = Field( - min_length=settings.PASSWORD_MIN_LENGTH, - max_length=settings.PASSWORD_MAX_LENGTH, + min_length = settings.PASSWORD_MIN_LENGTH, + max_length = settings.PASSWORD_MAX_LENGTH, ) @field_validator("password") @@ -30,11 +30,17 @@ class UserCreate(BaseModel): Validate password meets security requirements """ if not re.search(r"[A-Z]", v): - raise ValueError("Password must contain at least one uppercase letter") + raise ValueError( + "Password must contain at least one uppercase letter" + ) if not re.search(r"[a-z]", v): - raise ValueError("Password must contain at least one lowercase letter") + raise ValueError( + "Password must contain at least one lowercase letter" + ) if not re.search(r"[0-9]", v): - raise ValueError("Password must contain at least one number") + raise ValueError( + "Password must contain at least one number" + ) return v @@ -53,7 +59,7 @@ class UserResponse(BaseModel): Excludes sensitive fields like hashed_password. """ - model_config = ConfigDict(from_attributes=True) + model_config = ConfigDict(from_attributes = True) id: int email: str