crypto(inverted-oracle): add challenge

crypto/inverted-oracle/challenge.yml

@@ -0,0 +1,30 @@
+name: "Inverted Oracle"
+author: "rok0s"
+category: crypto
+
+description: |
+  Infiltrating the Orion Secure Terminal is the only path forward in uncovering the secrets of Project Echo
+  This high-security terminal, designed to control and monitor key aspects of the project, uses 
+  advanced cryptography to ensure that none of its secrets are leaked. Can you possible invert its security mechanisms?
+
+value: 500
+type: dynamic_docker
+extra:
+  initial: 500
+  minimum: 100
+  decay: 50
+  redirect_type: direct
+  compose_stack: !filecontents docker-compose.yml
+
+flags:
+  - CCSC{fL1p1Ty_Fl1P!!}
+
+tags:
+  - crypto
+  - medium
+
+files:
+  - "public/challenge.py"
+
+state: visible
+version: "0.1"

crypto/inverted-oracle/docker-compose.yml

@@ -0,0 +1,13 @@
+version: "3.7"
+
+services:
+  challenge:
+    image: ghcr.io/cybermouflons/ccsc2024/inverted-orable:latest
+    restart: always
+    ports:
+      - 13373:13373
+    build:
+      context: ./setup
+      dockerfile: Dockerfile
+      labels: 
+        ctf.challenge.name: inverted-oracle

crypto/inverted-oracle/public/challenge.py

@@ -0,0 +1,106 @@
+import math
+import os
+import string
+from functools import cache
+from itertools import cycle
+from typing import Tuple
+
+from banner import banner
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from secret import FLAG
+
+BLOCK_SIZE = 16
+KEY = os.urandom(32)
+
+
+@cache
+def get_random_blocks(n: int, block_size: int = BLOCK_SIZE) -> bytes:
+    if n > 1:
+        return get_random_blocks(n - 1) + os.urandom(block_size)
+    elif n == 1:
+        return os.urandom(block_size)
+    else:
+        return b""
+
+
+def xor(data: bytes, key: bytes) -> bytes:
+    return bytes(a ^ b for a, b in zip(data, cycle(key)))
+
+
+def encrypt(input_data: bytes, key: bytes) -> Tuple[bytes, bytes, bytes]:
+    iv = os.urandom(BLOCK_SIZE)
+    padder = padding.PKCS7(algorithms.AES.block_size).padder()
+    padded_data = padder.update(input_data) + padder.finalize()
+    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
+    encryptor = cipher.encryptor()
+    encrypted_data = encryptor.update(padded_data) + encryptor.finalize()
+    return encrypted_data, iv
+
+
+def decrypt(encrypted_data: bytes, key: bytes, iv: bytes) -> bytes:
+    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
+    decryptor = cipher.decryptor()
+    decrypted_data = decryptor.update(encrypted_data) + decryptor.finalize()
+    unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder()
+    unpadded_data = unpadder.update(decrypted_data) + unpadder.finalize()
+    return unpadded_data
+
+
+def start_new_session(key: bytes) -> tuple[bytes, bytes]:
+    user = input("Please enter a username: ")
+    user = "".join(c for c in user if c in string.ascii_lowercase)
+    session = f"user={user}&flag={FLAG}".encode()
+    enc_session, session_iv = encrypt(session, key)
+    return enc_session, session_iv
+
+
+def use_existing_session() -> bytes:
+    session_hex = input("Please enter an existing session (hex): ")
+    return bytes.fromhex(session_hex)
+
+
+def show_masked_flag(session: bytes, iv: bytes, key: bytes) -> None:
+    dec_cookie = decrypt(session, key, iv)
+    flag = dec_cookie.split(b"=")[-1]
+    flag_enc = xor(
+        flag,
+        get_random_blocks(math.ceil(len(flag) / BLOCK_SIZE)),
+    )
+    print("Masked flag:", flag_enc.hex())
+
+
+if __name__ == "__main__":
+    print(banner)
+    session, session_iv = None, None
+    while True:
+        try:
+            print("Please select one of the following options")
+            if session is None:
+                print("1. Start new session")
+                print("2. Use an existing encrypted session state")
+            else:
+                print("Current session:", session.hex())
+                print("1. Exit session")
+                print("2. Show masked flag")
+
+            option = input(">>> ")
+            if option not in ["1", "2"]:
+                print("Invalid option")
+                exit(1)
+
+            if session is None:
+                if option == "1":
+                    session, session_iv = start_new_session(KEY)
+                elif option == "2":
+                    session = use_existing_session()
+            else:
+                if option == "1":
+                    session = None
+                elif option == "2":
+                    show_masked_flag(session, session_iv, KEY)
+        except ValueError as e:
+            print("Invalid input", e)
+        except Exception as e:
+            print("An error occurred during decryption:", e)

crypto/inverted-oracle/setup/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.10-alpine
+
+RUN apk update
+RUN apk add socat gcc libc-dev libffi-dev
+
+WORKDIR /chall
+
+COPY . .
+
+RUN pip install poetry
+RUN poetry install
+
+EXPOSE 13373
+
+CMD ["socat", "-T60", "TCP-LISTEN:13373,reuseaddr,fork,su=root", "EXEC:python3 challenge.py"]

crypto/inverted-oracle/setup/banner.py

@@ -0,0 +1,15 @@
+banner = """******************************************************************
+*                                                                *
+*     ██████╗ ██████╗ ██╗ ██████╗ ███╗   ██╗ ████████╗ ███████╗  *
+*    ██╔═══██╗██╔══██╗██║██╔═══██╗████╗  ██║ ╚══██╔══╝ ██╔════╝  *
+*    ██║   ██║██████╔╝██║██║   ██║██╔██╗ ██║    ██║    ███████╗  *
+*    ██║   ██║██╔══██╗██║██║   ██║██║╚██╗██║    ██║    ╚════██║  *
+*    ╚██████╔╝██║  ██║██║╚██████╔╝██║ ╚████║    ██║    ███████║  *
+*     ╚═════╝ ╚═╝  ╚═╝╚═╝ ╚═════╝ ╚═╝  ╚═══╝    ╚═╝    ╚══════╝  *
+*                                                                *
+******************************************************************
+*                    Welcome to OrionTech's                      *
+*                     Secure Terminal Access                     *
+******************************************************************
+
+"""

crypto/inverted-oracle/setup/challenge.py

@@ -0,0 +1,106 @@
+import math
+import os
+import string
+from functools import cache
+from itertools import cycle
+from typing import Tuple
+
+from banner import banner
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from secret import FLAG
+
+BLOCK_SIZE = 16
+KEY = os.urandom(32)
+
+
+@cache
+def get_random_blocks(n: int, block_size: int = BLOCK_SIZE) -> bytes:
+    if n > 1:
+        return get_random_blocks(n - 1) + os.urandom(block_size)
+    elif n == 1:
+        return os.urandom(block_size)
+    else:
+        return b""
+
+
+def xor(data: bytes, key: bytes) -> bytes:
+    return bytes(a ^ b for a, b in zip(data, cycle(key)))
+
+
+def encrypt(input_data: bytes, key: bytes) -> Tuple[bytes, bytes, bytes]:
+    iv = os.urandom(BLOCK_SIZE)
+    padder = padding.PKCS7(algorithms.AES.block_size).padder()
+    padded_data = padder.update(input_data) + padder.finalize()
+    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
+    encryptor = cipher.encryptor()
+    encrypted_data = encryptor.update(padded_data) + encryptor.finalize()
+    return encrypted_data, iv
+
+
+def decrypt(encrypted_data: bytes, key: bytes, iv: bytes) -> bytes:
+    cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
+    decryptor = cipher.decryptor()
+    decrypted_data = decryptor.update(encrypted_data) + decryptor.finalize()
+    unpadder = padding.PKCS7(algorithms.AES.block_size).unpadder()
+    unpadded_data = unpadder.update(decrypted_data) + unpadder.finalize()
+    return unpadded_data
+
+
+def start_new_session(key: bytes) -> tuple[bytes, bytes]:
+    user = input("Please enter a username: ")
+    user = "".join(c for c in user if c in string.ascii_lowercase)
+    session = f"user={user}&flag={FLAG}".encode()
+    enc_session, session_iv = encrypt(session, key)
+    return enc_session, session_iv
+
+
+def use_existing_session() -> bytes:
+    session_hex = input("Please enter an existing session (hex): ")
+    return bytes.fromhex(session_hex)
+
+
+def show_masked_flag(session: bytes, iv: bytes, key: bytes) -> None:
+    dec_cookie = decrypt(session, key, iv)
+    flag = dec_cookie.split(b"=")[-1]
+    flag_enc = xor(
+        flag,
+        get_random_blocks(math.ceil(len(flag) / BLOCK_SIZE)),
+    )
+    print("Masked flag:", flag_enc.hex())
+
+
+if __name__ == "__main__":
+    print(banner)
+    session, session_iv = None, None
+    while True:
+        try:
+            print("Please select one of the following options")
+            if session is None:
+                print("1. Start new session")
+                print("2. Use an existing encrypted session state")
+            else:
+                print("Current session:", session.hex())
+                print("1. Exit session")
+                print("2. Show masked flag")
+
+            option = input(">>> ")
+            if option not in ["1", "2"]:
+                print("Invalid option")
+                exit(1)
+
+            if session is None:
+                if option == "1":
+                    session, session_iv = start_new_session(KEY)
+                elif option == "2":
+                    session = use_existing_session()
+            else:
+                if option == "1":
+                    session = None
+                elif option == "2":
+                    show_masked_flag(session, session_iv, KEY)
+        except ValueError as e:
+            print("Invalid input", e)
+        except Exception as e:
+            print("An error occurred during decryption:", e)