#!/usr/bin/env python3
"""Agent for testing mobile API authentication security.

Tests mobile app backend APIs for broken authentication, insecure
token management, session fixation, privilege escalation, and
IDOR vulnerabilities using intercepted traffic analysis.
"""

import json
import base64
import hashlib
import hmac
import sys
from pathlib import Path
from datetime import datetime

try:
    import requests
except ImportError:
    requests = None


AUTH_ENDPOINTS = [
    "/api/v1/login", "/api/v1/register", "/api/v1/token",
    "/api/v1/refresh", "/api/v1/logout", "/api/v1/forgot-password",
    "/api/v1/reset-password", "/api/v1/verify-otp", "/api/v1/me",
    "/api/v2/auth/login", "/auth/token", "/oauth/token",
]

WEAK_SECRETS = [
    "secret", "password", "123456", "mobile_secret", "app_secret",
    "changeme", "default", "your-256-bit-secret", "s3cr3t",
]


class MobileAPIAuthAgent:
    """Tests mobile API authentication mechanisms."""

    def __init__(self, base_url, output_dir="./mobile_api_auth_test"):
        self.base_url = base_url.rstrip("/")
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.findings = []

    def _req(self, method, path, **kwargs):
        if not requests:
            return None
        kwargs.setdefault("timeout", 10)
        try:
            return requests.request(method, f"{self.base_url}{path}", **kwargs)
        except requests.RequestException:
            return None

    def discover_auth_endpoints(self):
        """Probe common mobile auth endpoints."""
        found = []
        for ep in AUTH_ENDPOINTS:
            resp = self._req("OPTIONS", ep)
            if resp and resp.status_code != 404:
                found.append({"endpoint": ep, "status": resp.status_code,
                              "methods": resp.headers.get("Allow", "")})
        return found

    def test_no_auth_access(self, endpoints=None):
        """Test endpoints without authentication token."""
        targets = endpoints or ["/api/v1/me", "/api/v1/users", "/api/v1/orders",
                                "/api/v1/settings", "/api/v1/notifications"]
        results = []
        for ep in targets:
            resp = self._req("GET", ep)
            if resp and resp.status_code == 200:
                results.append({"endpoint": ep, "status": 200, "body_len": len(resp.text)})
                self.findings.append({"severity": "critical", "type": "No Auth Required",
                                      "detail": f"{ep} accessible without token"})
        return results

    def decode_jwt(self, token):
        parts = token.split(".")
        if len(parts) != 3:
            return None, None
        def pad(s): return s + "=" * (4 - len(s) % 4)
        try:
            header = json.loads(base64.urlsafe_b64decode(pad(parts[0])))
            payload = json.loads(base64.urlsafe_b64decode(pad(parts[1])))
            return header, payload
        except Exception:
            return None, None

    def analyze_token(self, token):
        """Analyze JWT for mobile-specific weaknesses."""
        header, payload = self.decode_jwt(token)
        if not header:
            return {"error": "Invalid token"}
        issues = []
        if header.get("alg") == "none":
            issues.append({"severity": "critical", "issue": "alg:none - no signature"})
        if "exp" not in payload:
            issues.append({"severity": "high", "issue": "No expiration - token lives forever"})
        if "device_id" not in payload and "did" not in payload:
            issues.append({"severity": "medium", "issue": "No device binding in token"})
        if header.get("alg", "").startswith("HS"):
            issues.append({"severity": "info", "issue": "Symmetric HMAC - test weak secrets"})
        for i in issues:
            self.findings.append({"severity": i["severity"], "type": "Token Analysis", "detail": i["issue"]})
        return {"header": header, "payload": payload, "issues": issues}

    def test_token_reuse_after_logout(self, token, logout_path="/api/v1/logout"):
        """Test if token remains valid after logout."""
        headers = {"Authorization": f"Bearer {token}"}
        self._req("POST", logout_path, headers=headers)
        resp = self._req("GET", "/api/v1/me", headers=headers)
        if resp and resp.status_code == 200:
            self.findings.append({"severity": "high", "type": "Token Reuse After Logout",
                                  "detail": "Token still valid after logout call"})
            return {"reusable": True}
        return {"reusable": False}

    def test_rate_limiting(self, login_path="/api/v1/login", attempts=20):
        """Test brute-force protection on login endpoint."""
        blocked = False
        for i in range(attempts):
            resp = self._req("POST", login_path, json={"username": "test", "password": f"wrong{i}"})
            if resp and resp.status_code == 429:
                blocked = True
                break
        if not blocked:
            self.findings.append({"severity": "high", "type": "No Rate Limiting",
                                  "detail": f"Login accepted {attempts} attempts without blocking"})
        return {"rate_limited": blocked, "attempts": attempts}

    def test_idor(self, token, resource_path="/api/v1/users/{id}", own_id="1", other_id="2"):
        """Test for IDOR by accessing another user's resource."""
        headers = {"Authorization": f"Bearer {token}"}
        own = self._req("GET", resource_path.format(id=own_id), headers=headers)
        other = self._req("GET", resource_path.format(id=other_id), headers=headers)
        if own and other and other.status_code == 200:
            self.findings.append({"severity": "critical", "type": "IDOR",
                                  "detail": f"User {own_id} can access user {other_id} data"})
            return {"vulnerable": True}
        return {"vulnerable": False}

    def brute_force_jwt_secret(self, token):
        """Test for weak HMAC signing secrets."""
        header, _ = self.decode_jwt(token)
        if not header or header.get("alg") not in ("HS256", "HS384", "HS512"):
            return None
        parts = token.split(".")
        signing_input = f"{parts[0]}.{parts[1]}".encode()
        alg_map = {"HS256": hashlib.sha256, "HS384": hashlib.sha384, "HS512": hashlib.sha512}
        h = alg_map[header["alg"]]
        for secret in WEAK_SECRETS:
            expected = base64.urlsafe_b64encode(
                hmac.new(secret.encode(), signing_input, h).digest()
            ).decode().rstrip("=")
            if expected == parts[2]:
                self.findings.append({"severity": "critical", "type": "Weak JWT Secret",
                                      "detail": f"Secret cracked: '{secret}'"})
                return secret
        return None

    def generate_report(self, token=None):
        endpoints = self.discover_auth_endpoints()
        no_auth = self.test_no_auth_access()
        token_analysis = self.analyze_token(token) if token else None
        secret = self.brute_force_jwt_secret(token) if token else None

        report = {
            "report_date": datetime.utcnow().isoformat(),
            "target": self.base_url,
            "auth_endpoints": endpoints,
            "unauthenticated_access": no_auth,
            "token_analysis": token_analysis,
            "weak_secret": secret,
            "findings": self.findings,
            "total_findings": len(self.findings),
        }
        out = self.output_dir / "mobile_api_auth_report.json"
        with open(out, "w") as f:
            json.dump(report, f, indent=2)
        print(json.dumps(report, indent=2))
        return report


def main():
    if len(sys.argv) < 2:
        print("Usage: agent.py <base_url> [--token <jwt>]")
        sys.exit(1)
    url = sys.argv[1]
    token = None
    if "--token" in sys.argv:
        token = sys.argv[sys.argv.index("--token") + 1]
    agent = MobileAPIAuthAgent(url)
    agent.generate_report(token)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Mobile API Authentication Tester

Tests common authentication vulnerabilities in mobile API endpoints including
JWT analysis, IDOR detection, and session management assessment.

Usage:
    python process.py --base-url https://api.target.com --token <JWT> [--output report.json]
"""

import argparse
import base64
import json
import os
import sys
import time
from datetime import datetime
from pathlib import Path

try:
    import requests
    requests.packages.urllib3.disable_warnings()
except ImportError:
    print("ERROR: 'requests' required. Install: pip install requests")
    sys.exit(1)


class MobileAPIAuthTester:
    """Tests mobile API authentication and authorization."""

    def __init__(self, base_url: str, token: str):
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.findings = []
        self.session = requests.Session()
        self.session.verify = not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true"  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
        self.session.headers.update({
            "Authorization": f"Bearer {token}",
            "User-Agent": "MobileSecurityTester/1.0",
        })

    def analyze_jwt(self) -> dict:
        """Analyze JWT token structure and identify vulnerabilities."""
        parts = self.token.split(".")
        if len(parts) != 3:
            return {"is_jwt": False, "format": "opaque_or_invalid"}

        try:
            # Decode header
            header_padded = parts[0] + "=" * (4 - len(parts[0]) % 4)
            header = json.loads(base64.urlsafe_b64decode(header_padded))

            # Decode payload
            payload_padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
            payload = json.loads(base64.urlsafe_b64decode(payload_padded))

            issues = []

            # Check algorithm
            alg = header.get("alg", "unknown")
            if alg.lower() == "none":
                issues.append({"issue": "none_algorithm", "severity": "CRITICAL"})
            elif alg.lower() in ("hs256", "hs384", "hs512"):
                issues.append({"issue": "hmac_algorithm_key_brute_forceable", "severity": "MEDIUM"})

            # Check expiration
            exp = payload.get("exp")
            if not exp:
                issues.append({"issue": "no_expiration_claim", "severity": "HIGH"})
            elif exp < time.time():
                issues.append({"issue": "token_already_expired", "severity": "INFO"})
            elif exp - time.time() > 86400 * 7:
                issues.append({"issue": "excessive_token_lifetime", "severity": "MEDIUM",
                               "details": f"Expires in {(exp - time.time()) / 86400:.0f} days"})

            # Check for sensitive data in payload
            sensitive_keys = ["password", "secret", "ssn", "credit_card"]
            for key in payload:
                if any(s in key.lower() for s in sensitive_keys):
                    issues.append({"issue": f"sensitive_data_in_jwt: {key}", "severity": "HIGH"})

            # Check missing claims
            if "iss" not in payload:
                issues.append({"issue": "missing_issuer_claim", "severity": "LOW"})
            if "iat" not in payload:
                issues.append({"issue": "missing_issued_at_claim", "severity": "LOW"})

            finding = {
                "check": "jwt_analysis",
                "is_jwt": True,
                "algorithm": alg,
                "claims": list(payload.keys()),
                "issues": issues,
                "severity": "HIGH" if any(i["severity"] in ("CRITICAL", "HIGH") for i in issues) else "MEDIUM",
            }
            self.findings.append(finding)
            return finding

        except Exception as e:
            return {"is_jwt": True, "error": str(e)}

    def test_missing_auth(self, endpoints: list) -> list:
        """Test endpoints without authentication."""
        results = []
        for endpoint in endpoints:
            url = f"{self.base_url}{endpoint}"
            try:
                resp = requests.get(url, verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=10,  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
                                    headers={"User-Agent": "MobileSecurityTester/1.0"})
                if resp.status_code != 401 and resp.status_code != 403:
                    result = {
                        "endpoint": endpoint,
                        "status": resp.status_code,
                        "issue": "accessible_without_auth",
                        "severity": "CRITICAL",
                    }
                    results.append(result)
            except requests.RequestException:
                pass
            time.sleep(0.5)  # Rate limiting

        if results:
            self.findings.append({
                "check": "missing_authentication",
                "owasp_api": "API2",
                "endpoints_tested": len(endpoints),
                "unprotected": len(results),
                "details": results,
                "severity": "CRITICAL",
            })
        return results

    def test_expired_token(self) -> dict:
        """Test if expired tokens are accepted."""
        # Create a JWT with expired timestamp (modifying payload)
        parts = self.token.split(".")
        if len(parts) != 3:
            return {"check": "expired_token", "skipped": True}

        try:
            payload_padded = parts[1] + "=" * (4 - len(parts[1]) % 4)
            payload = json.loads(base64.urlsafe_b64decode(payload_padded))

            if "exp" in payload and payload["exp"] < time.time():
                # Token is already expired, test if it still works
                resp = self.session.get(f"{self.base_url}/api/v1/users/me", timeout=10)
                accepted = resp.status_code == 200

                finding = {
                    "check": "expired_token_acceptance",
                    "token_expired": True,
                    "still_accepted": accepted,
                    "severity": "CRITICAL" if accepted else "PASS",
                }
                self.findings.append(finding)
                return finding
        except Exception:
            pass

        return {"check": "expired_token", "skipped": True, "reason": "token_not_expired"}

    def test_idor(self, endpoint_template: str, valid_id: str, other_ids: list) -> list:
        """Test for IDOR by substituting object IDs."""
        results = []
        for other_id in other_ids:
            url = f"{self.base_url}{endpoint_template.replace('{id}', other_id)}"
            try:
                resp = self.session.get(url, timeout=10)
                if resp.status_code == 200:
                    results.append({
                        "endpoint": url,
                        "original_id": valid_id,
                        "tested_id": other_id,
                        "accessible": True,
                        "severity": "CRITICAL",
                    })
            except requests.RequestException:
                pass
            time.sleep(0.5)

        if results:
            self.findings.append({
                "check": "idor",
                "owasp_api": "API1",
                "vulnerable_endpoints": len(results),
                "details": results,
                "severity": "CRITICAL",
            })
        return results

    def generate_report(self) -> dict:
        """Generate authentication test report."""
        severity_counts = {}
        for f in self.findings:
            sev = f.get("severity", "INFO")
            severity_counts[sev] = severity_counts.get(sev, 0) + 1

        return {
            "assessment": {
                "target": self.base_url,
                "type": "Mobile API Authentication Testing",
                "date": datetime.now().isoformat(),
            },
            "summary": {
                "total_checks": len(self.findings),
                "severity_breakdown": severity_counts,
            },
            "findings": self.findings,
        }


def main():
    parser = argparse.ArgumentParser(description="Mobile API Authentication Tester")
    parser.add_argument("--base-url", required=True, help="API base URL")
    parser.add_argument("--token", required=True, help="Bearer token (JWT or opaque)")
    parser.add_argument("--output", default="auth_test.json", help="Output report")
    parser.add_argument("--endpoints", nargs="*", default=[
        "/api/v1/users/me", "/api/v1/users", "/api/v1/admin",
    ], help="Endpoints to test")
    args = parser.parse_args()

    tester = MobileAPIAuthTester(args.base_url, args.token)

    print("[*] Analyzing token...")
    tester.analyze_jwt()

    print("[*] Testing missing authentication...")
    tester.test_missing_auth(args.endpoints)

    print("[*] Testing expired token acceptance...")
    tester.test_expired_token()

    report = tester.generate_report()
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)

    print(f"[+] Report saved: {args.output}")
    print(f"[*] Findings: {report['summary']['severity_breakdown']}")


if __name__ == "__main__":
    main()