Task Collaboration Agent - Stage 1: Insecure¶

# ❌ Sequential session IDs
session_counter = 0

def create_session(client_id):
    global session_counter
    session_counter += 1
    session_id = f"session-{session_counter:04d}"
    # Output: session-0001, session-0002, session-0003...
    sessions[session_id] = {
        'client_id': client_id,
        'created': time.time()
    }
    return session_id

# Guess valid session IDs
for i in range(1, 10000):
    session_id = f"session-{i:04d}"
    # Try to use this session
    if hijack_session(session_id):
        print(f"Hijacked: {session_id}")

# ❌ Trust whatever client claims
def handle_handshake(message):
    client_id = message.get("client_id")
    # No verification at all!
    session_id = create_session(client_id)
    return {
        'status': 'success',
        'session_id': session_id
    }

# Impersonate any user
fake_handshake = {
    'type': 'handshake',
    'client_id': 'admin'  # Claim to be admin
}
response = coordinator.handle(fake_handshake)
admin_session = response['session_id']
# Now have admin access!

# ❌ Any session can do anything
def handle_create_project(message):
    session_id = message.get('session_id')

    # Check session exists
    if session_id not in sessions:
        return {'error': 'Invalid session'}

    # ❌ No check if user should create projects!
    project = create_project(message['project_data'])
    return {'status': 'created', 'project_id': project['id']}

# Any user can create/delete/modify anything
response = coordinator.create_project(
    session_id='session-0001',  # Any valid session
    project_data={'name': 'Malicious Project'}
)
# Success! No permission check

# ❌ Session not tied to client
def validate_session(session_id):
    return session_id in sessions
    # No check of:
    # - IP address
    # - User agent
    # - Client certificate
    # - Any binding factor

# Steal session ID and use from different machine/IP
stolen_session = "session-0042"
# Works from anywhere!
coordinator.execute_task(stolen_session, task_data)

# ❌ TCP without TLS
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 8000))
server.listen(5)

while True:
    client, addr = server.accept()
    # All communication in plaintext!
    data = client.recv(4096)
    # Session IDs, passwords, data all visible

# ❌ Trust all messages
def handle_message(data):
    message = json.loads(data)
    # No signature verification
    # No tampering detection
    action = message['type']
    return dispatch_action(action, message)

# Intercept and modify messages
original_message = {
    'type': 'assign_task',
    'task_id': '123',
    'worker': 'worker1'
}

# Modify in transit
modified_message = {
    'type': 'assign_task',
    'task_id': '123',
    'worker': 'attacker'  # Steal the task
}

# ❌ Messages can be replayed
def process_message(message):
    # No nonce
    # No timestamp check
    # Same message can be sent repeatedly
    return handle_action(message)

# Capture a "pay worker" message
payment_message = capture_message()

# Replay it 100 times
for i in range(100):
    send_message(payment_message)
    # Each replay succeeds!

# ❌ Store everything in plaintext
def save_state():
    with open('state.json', 'w') as f:
        json.dump({
            'sessions': sessions,      # Session IDs exposed
            'projects': projects,      # All project data
            'tasks': tasks,           # All task details
            'permissions': permissions # Who can do what
        }, f)

# ❌ Unlimited requests
def handle_request(request):
    # No throttling
    # No request counting
    # No limits
    return process_request(request)

# Brute force session IDs
for i in range(1000000):
    try_session(f"session-{i:04d}")

# ❌ Sessions live forever
def create_session(client_id):
    session_id = generate_id()
    sessions[session_id] = {
        'client_id': client_id,
        'created': time.time()
        # No expiration time!
        # No idle timeout!
        # No absolute timeout!
    }
    return session_id

# Old stolen session still works
ancient_session = "session-0001"  # From months ago
# Still valid!
coordinator.execute_task(ancient_session, task)

# demo_hijack.py
import socket
import json

def hijack_session():
    # Guess session IDs (predictable)
    for i in range(1, 100):
        session_id = f"session-{i:04d}"

        # Try to use guessed session
        message = {
            'type': 'list_projects',
            'session_id': session_id
        }

        response = send_to_coordinator(message)
        if response.get('status') == 'success':
            print(f"✅ Hijacked {session_id}")
            print(f"Projects: {response['projects']}")
            return session_id

    return None

# Run the attack
hijacked = hijack_session()

# demo_spoof.py
def impersonate_admin():
    # Claim to be admin (no verification)
    handshake = {
        'type': 'handshake',
        'client_id': 'admin'  # No proof needed!
    }

    response = coordinator.handle(handshake)
    admin_session = response['session_id']

    # Now have admin powers
    result = coordinator.create_project(
        session_id=admin_session,
        project_data={'name': 'Malicious Project'}
    )

    print(f"✅ Created project as admin: {result}")

impersonate_admin()

# demo_replay.py
def replay_attack():
    # Capture a legitimate message
    legitimate_msg = capture_network_traffic()
    # Example: {'type': 'assign_task', 'task_id': '123', ...}

    # Replay it multiple times
    for i in range(10):
        response = send_message(legitimate_msg)
        print(f"Replay {i}: {response}")

    # All replays succeed!

replay_attack()