MythEclipse/attack-defense-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

removed old files

Browse Source
This commit is contained in:
Rayhan Hanaputra
2025-10-11 11:59:21 +07:00
parent 9b1227378d
commit d42b472b3e
1230 changed files with 28 additions and 288310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
receiver/challenges.old/Art.py
-40
View File
@@ -1,40 +0,0 @@
from .Challenge import Challenge
import io
import pandas as pd
import requests
import re
class Art(Challenge):
flag_location = 'flags/art.txt'
history_location = 'history/art.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
word = self.random_string(8)
url = f'http://localhost:{self.port}/art/{word}'
r = requests.get(url, timeout=5)
assert r.text == f'<iframe height="100%" width="100%" frameborder="0" src=https://asciified.thelicato.io/api/v2/ascii?text={word}></iframe>', 'Unexpected response'
self.logger.info('Check passed for art')
return True
except Exception as e:
self.logger.error(f'Could not check art: {e}')
return False
receiver/challenges.old/BackToBasic.py
-35
View File
@@ -1,35 +0,0 @@
from .Challenge import Challenge
from pwn import *
class BackToBasic(Challenge):
flag_location = 'flags/back-to-basic.txt'
history_location = 'history/back-to-basic.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
r = remote("localhost",self.port)
assert b"idea?" in r.recvline(), "Failed First"
r.sendline(b"testt")
assert b"thing" in r.recvline(), "Failed Last"
return True
except Exception as e:
self.logger.error(f'Could not check back-to-basic: {e}')
return False
receiver/challenges.old/Burvesigner.py
-124
View File
@@ -1,124 +0,0 @@
from .Challenge import Challenge
from fastecdsa.curve import Curve
from fastecdsa.point import Point
import requests
import time
import os
import json
class Burvesigner(Challenge):
flag_location = 'flags/burvesigner.txt'
history_location = 'history/burvesigner.txt'
priv_location = 'files/burvesigner.priv'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.priv_location, 'wb') as f:
f.write(os.urandom(256))
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
url = f'http://localhost:{self.port}'
flag = open(self.flag_location).read()
# C1: login guest success
guest_data = {
'username': 'guest',
'password': 'guest',
}
response = requests.post(url, data=guest_data, timeout=5)
guest_token = response.cookies["token"]
assert "Welcome, guest!" in response.text, "Guest cannot login"
assert "flashes" not in response.text, "Guest cannot login"
self.logger.info(f'C1 success for burvesigner')
# C2: login admin success
admin_data = {
'username': 'merricx_number_1_fans',
'password': getattr(self.settings, f'PASSWORD_{self.port}'),
}
response = requests.post(url, data=admin_data, timeout=5)
admin_token = response.cookies["token"]
assert "Welcome, merricx_number_1_fans!" in response.text, "Admin cannot login"
assert flag in response.text, "Flag is missing in admin page"
assert "flashes" not in response.text, "Admin cannot login"
self.logger.info(f'C2 success for burvesigner')
# C3: login guest fail
guest_data = {
'username': 'guest',
'password': 'Guest',
}
response = requests.post(url, data=guest_data, timeout=5).text
assert "Welcome, guest!" not in response, "Guest with wrong credential can login"
assert "flashes" in response, "Guest with wrong credential can login"
self.logger.info(f'C3 success for burvesigner')
# C4: login admin fail
admin_data = {
'username': 'merricx_number_1_fans',
'password': 'password',
}
response = requests.post(url, data=admin_data, timeout=5).text
assert "Welcome, merricx_number_1_fans!" not in response, "Admin with wrong credential can login"
assert flag not in response, "Flag is showing in admin page with wrong credential"
assert "flashes" in response, "Admin with wrong credential can login"
self.logger.info(f'C4 success for burvesigner')
self.logger.info('calling sleep(5)')
time.sleep(5)
# C5: cek apakah token guest di C1 sudah expired atau belum
response = requests.get(url, cookies={"token": guest_token}, timeout=5)
assert "Welcome, guest!" in response.text, "Guest token expires early"
assert "flashes" not in response.text, "Guest token expires early"
self.logger.info(f'C5 success for burvesigner')
# C6: cek apakah token admin di C2 sudah expired atau belum
response = requests.get(url, cookies={"token": admin_token}, timeout=5)
assert "Welcome, merricx_number_1_fans!" in response.text, "Admin token expires early"
assert flag in response.text, "Admin token expires early"
assert "flashes" not in response.text, "Admin token expires early"
self.logger.info(f'C6 success for burvesigner')
# C7: cek endpoint /params
response = requests.get(url + "/params", timeout=5).text
response = response.replace("<pre>", "").replace("</pre>", "")
params = json.loads(response)
assert params["p"] and params["a"] and params["b"] and params["n"], "Missing p, a, b and/or n parameter(s)"
assert params["G"][0] and params["G"][1] and params["Y"][0] and params["Y"][1], "Missing G and/or Y point(s)"
self.logger.info(f'C7 success for burvesigner')
# C8: cek apakah curve C valid dan point G di C
C = Curve("burvesigner", params["p"], params["a"], params["b"], params["n"], params["G"][0], params["G"][1])
assert C.G == Point(params["G"][0], params["G"][1], C), "Point G is not valid"
self.logger.info(f'C8 success for burvesigner')
# C9: cek apakah point G * priv = Y
t = params["p"].bit_length() // 8
priv = open(self.priv_location, "rb").read()[:t]
x = int.from_bytes(priv, "little")
Y = Point(params["Y"][0], params["Y"][1], C)
assert C.G * x == Y, "Point Y is not valid"
self.logger.info(f'C9 success for burvesigner')
return True
except Exception as e:
self.logger.error(f'Could not check burvesigner: {e}')
return False
receiver/challenges.old/Challenge.py
-34
View File
@@ -1,34 +0,0 @@
import logging
import random
import string
from config import get_settings
class Challenge(object):
name = __name__
settings = get_settings()
port = 0
def __init__(self, port):
self.port = port
self.add_logger()
def add_logger(self):
self.logger = logging.getLogger()
def random_string(self, length):
charset = string.ascii_uppercase + string.ascii_lowercase + string.digits
return ''.join(random.choice(charset) for i in range(length))
def distribute(self, flag):
raise NotImplementedError
def check(self):
raise NotImplementedError
def credentials(self):
return {
'username': 'root',
'password': getattr(self.settings, f'PASSWORD_{self.port}'),
}
receiver/challenges.old/Crawlback.py
-38
View File
@@ -1,38 +0,0 @@
from .Challenge import Challenge
import requests
import os
MOCK_URL = 'http://google.com'
MOCK_DATA = '<HTML><HEAD><meta http-equiv="content-type" content="text/html;charset=utf-8">'
class Crawlback(Challenge):
flag_location = 'flags/crawlback.txt'
history_location = 'history/crawlback.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
r = requests.post(f"http://localhost:{self.port}/crawlback.php", data={'url': MOCK_URL})
assert r.text.split('\n').pop(0) == MOCK_DATA
return True
except Exception as e:
self.logger.error(f'Could not check crawlback: {e}')
return False
receiver/challenges.old/GemasFetcher.py
-67
View File
@@ -1,67 +0,0 @@
from .Challenge import Challenge
import requests
import zlib
import gzip
import json
MOCK_URL = 'http://google.com'
MOCK_DATA_WGET = 'Google</title>'
MOCK_DATA_CURL = '<HTML><HEAD><meta http-equiv="content-type" content="text/html;charset=utf-8">'
class GemasFetcher(Challenge):
flag_location = 'flags/gemas-fetcher.txt'
history_location = 'history/gemas-fetcher.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
sess = requests.Session()
## register
username = self.random_string(5)
password = self.random_string(5)
r = sess.post(f"http://localhost:{self.port}/auth/register", data={"username":username,"password": password}, allow_redirects=False)
assert r.headers.get("location") == "/auth/login", "Register Failed"
## login
r = sess.post(f"http://localhost:{self.port}/auth/login", data={"username":username,"password": password}, allow_redirects=False)
assert r.headers.get("location") == "/dashboard", "Login Failed"
## wget
content = {"provider": "wget","url":MOCK_URL}
files = {"file": ("visit", b"\x00\x00"+gzip.compress(zlib.compress(json.dumps(content).encode())))}
r = sess.post(f"http://localhost:{self.port}/dashboard/fetch_by_file", files=files)
assert MOCK_DATA_WGET in r.text, "wget Failed"
## curl
content = {"provider": "curl","url":MOCK_URL}
files = {"file": ("visit", b"\x00\x01"+gzip.compress(zlib.compress(json.dumps(content).encode())))}
r = sess.post(f"http://localhost:{self.port}/dashboard/fetch_by_file", files=files)
assert r.text.split('\n').pop(0) == MOCK_DATA_CURL, "curl Failed"
## python
content = {"provider": "python","url":MOCK_URL}
files = {"file": ("visit", b"\x00\x02"+gzip.compress(zlib.compress(json.dumps(content).encode())))}
r = sess.post(f"http://localhost:{self.port}/dashboard/fetch_by_file", files=files)
assert r.text.startswith('"PCFkb2N0eXBlIGh0bWw'), "python Failed"
return True
except Exception as e:
self.logger.error(f'Could not check gemas-fetcher: {e}')
return False
receiver/challenges.old/GemasNotes.py
-67
View File
@@ -1,67 +0,0 @@
from .Challenge import Challenge
import requests
class GemasNotes(Challenge):
history_location = 'history/gemas-notes.txt'
host = "http://localhost:12000"
def distribute(self, flag):
try:
username = "gemasflagreceiver"
password = "AuTeEbn%.Q5$pC_ge6"
result = requests.post(f"{self.host}/flag_receiver", json={"flag": flag}, auth=(username,password)).json()
if not result.get("success"):
return False
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} updated in gemas-notes database')
return True
except Exception as e:
self.logger.error(f'Could send flag to gemas-notes challenge: {e}')
return False
def check(self):
try:
url = f'http://localhost:{self.port}'
# login
token = requests.post(f"{url}/api/login",json={"email":"checker@gemasnotes.id", "password":"uRIqCvJ<IGb;VDT14"}).json()["token"]
header = {"Authorization": f"Bearer {token}"}
# get count
old_count = requests.post(f"{url}/api/notes/count", headers=header, json={"count_by":"title", "keyword":""}).json()["count"]
# create notes
notes = {"title":self.random_string(10), "content":self.random_string(20), "tags":self.random_string(10)}
status_code = requests.put(f"{url}/api/notes", headers=header, json=notes).status_code
assert status_code in [200, 201], "Cannot Create Note"
# get notes
all_notes = requests.get(f"{url}/api/notes").json()
note = list(filter(lambda x: x["title"] == notes["title"], all_notes))
assert len(note) != 0, "Note was not created"
# get new count
new_count = requests.post(f"{url}/api/notes/count", headers=header, json={"count_by":"title", "keyword":""}).json()["count"]
assert old_count != new_count, "Invalid count"
# update notes
new_content = self.random_string(20)
notes["id"] = note[0]["id"]
notes["content"] = new_content
status_code = requests.patch(f"{url}/api/notes", headers=header, json=notes).status_code
assert status_code in [200, 204], "Cannot Update Note"
# delete notes
status_code = requests.delete(f"{url}/api/notes/{notes['id']}", headers=header, json=notes).status_code
assert status_code == 200, "Cannot Delete Note"
return True
except Exception as e:
self.logger.error(f'Could not check gemas-notes: {e}')
return False
receiver/challenges.old/Hirnfick.py
-42
View File
@@ -1,42 +0,0 @@
import requests
from base64 import b64decode
from .Challenge import Challenge
class Hirnfick(Challenge):
flag_location = 'flags/hirnfick.txt'
history_location = 'history/hirnfick.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(
f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
res = requests.post(
f"http://localhost:{self.port}/api/run",
timeout=5,
json={
"code":
"+[-->-[>>+>-----<<]<--<---]>-.>>>+.>>..+++[.>]<<<<.+++.------.<<-.>>>>+."
})
assert b64decode(res.json()["output"]) == b"HirnFick 1.0\nHello, World!"
return True
except Exception as e:
self.logger.error(f'Could not check hirnfick: {e}')
return False
receiver/challenges.old/Pasta.py
-109
View File
@@ -1,109 +0,0 @@
from .Challenge import Challenge
import requests
class Pasta(Challenge):
flag_location = 'flags/pasta.txt'
history_location = 'history/pasta.txt'
host = "http://localhost:13000"
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
url = f'http://localhost:{self.port}'
username = f"checker-{self.random_string(8)}"
pwd = self.random_string(12)
flag = open(self.flag_location).read()
admin_data = {
'username': 'deomkicer_number_1_fans',
'password': getattr(self.settings, f'PASSWORD_{self.port}'),
}
# login admin and check flag
response = requests.post(
f"{url}/auth",
json=admin_data).json()
token = response.get('token')
assert token, "Token is missing in login admin"
check_flag = requests.get(f"{url}/flag", headers={'Authorization': f"Bearer {token}"}).json()
assert check_flag.get('flag') == flag, "Flag is missing/mismatch"
# register
response = requests.post(
f"{url}/register",
json={
"username": f"{username}",
"password": f"{pwd}"}).json()
assert response.get('success') == "User registered succesfully", "Register failed"
# login with version 1
response = requests.post(
f"{url}/auth?version=1",
json={
"username": f"{username}",
"password": f"{pwd}"}).json()
token = response.get('token')
assert token, "Token is missing in login v1"
check_home = requests.get(f"{url}/", headers={'Authorization': f"Bearer {token}"}).json()
assert check_home.get('username') == username, "Different username found in login v1"
# login with version 2
response = requests.post(
f"{url}/auth?version=2",
json={
"username": f"{username}",
"password": f"{pwd}"}).json()
token = response.get('token')
assert token, "Token is missing in login v2"
check_home = requests.get(f"{url}/", headers={'Authorization': f"Bearer {token}"}).json()
assert check_home.get('username') == username, "Different username found in login v2"
# login with version 3
response = requests.post(
f"{url}/auth?version=3",
json={
"username": f"{username}",
"password": f"{pwd}"}).json()
token = response.get('token')
assert token, "Token is missing in login v3"
check_home = requests.get(f"{url}/", headers={'Authorization': f"Bearer {token}"}).json()
assert check_home.get('username') == username, "Different username found in login v3"
# login with version 4
response = requests.post(
f"{url}/auth?version=4",
json={
"username": f"{username}",
"password": f"{pwd}"}).json()
token = response.get('token')
assert token, "Token is missing in login v4"
check_home = requests.get(f"{url}/", headers={'Authorization': f"Bearer {token}"}).json()
assert check_home.get('username') == username, "Different username found in login v4"
return True
except Exception as e:
self.logger.error(f'Could not check pasta: {e}')
return False
receiver/challenges.old/S3.py
-44
View File
@@ -1,44 +0,0 @@
from .Challenge import Challenge
import requests
import os
class S3(Challenge):
flag_location = 'flags/s3.txt'
history_location = 'history/s3.txt'
host = 'http://localhost:20000'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
filename = self.random_string(8) + ".txt"
content = self.random_string(64)
r = requests.post(f"http://localhost:{self.port}/upload", files={'file': (filename, content)})
assert r.status_code == 200
assert r.text == f'Download <a href="/download?filename={filename}">here</a>'
r = requests.get(f"http://localhost:{self.port}/download?filename={filename}")
assert r.status_code == 200
assert r.text == content
return True
except Exception as e:
self.logger.error(f'Could not check s3: {e}')
return False
receiver/challenges.old/XL.py
-66
View File
@@ -1,66 +0,0 @@
from .Challenge import Challenge
import io
import pandas as pd
import requests
import re
MOCK_DATA = [
{'name': 'John','age': 30, 'city': 'New York'},
{'name': 'Mary', 'age': 25, 'city': 'San Francisco'},
{'name': 'Peter', 'age': 45, 'city': 'Chicago'},
]
MOCK_RESULT = {
"Sheet1":{
"!ref":"A1:C4",
"A1":{"t":"s","v":"name","h":"name","w":"name"},"B1":{"t":"s","v":"age","h":"age","w":"age"},"C1":{"t":"s","v":"city","h":"city","w":"city"},
"A2":{"t":"s","v":"John","h":"John","w":"John"},"B2":{"t":"n","v":30,"w":"30"},"C2":{"t":"s","v":"New York","h":"New York","w":"New York"},
"A3":{"t":"s","v":"Mary","h":"Mary","w":"Mary"},"B3":{"t":"n","v":25,"w":"25"},"C3":{"t":"s","v":"San Francisco","h":"San Francisco","w":"San Francisco"},
"A4":{"t":"s","v":"Peter","h":"Peter","w":"Peter"},"B4":{"t":"n","v":45,"w":"45"},"C4":{"t":"s","v":"Chicago","h":"Chicago","w":"Chicago"},
"!margins":{"left":0.75,"right":0.75,"top":1,"bottom":1,"header":0.5,"footer":0.5}
}
}
class XL(Challenge):
flag_location = 'flags/xl.txt'
history_location = 'history/xl.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
url = f'http://localhost:{self.port}'
files = {'file': self.generate_mock_file()}
r = requests.post(url, files=files, timeout=5)
assert r.json() == MOCK_RESULT, 'Unexpected response'
self.logger.info('Check passed for xl')
return True
except Exception as e:
self.logger.error(f'Could not check xl: {e}')
return False
def generate_mock_file(self):
memory_file = io.BytesIO()
df = pd.DataFrame(MOCK_DATA)
df.to_excel(memory_file, index=False)
memory_file.seek(0)
return memory_file
receiver/challenges/Blinkpdf.py
-105
View File
@@ -1,105 +0,0 @@
from .Challenge import Challenge
from modules.blinkpdf import *
import io
import requests
import subprocess
import re
class BlinkPDF(Challenge):
flag_location = 'flags/blinkpdf.txt'
history_location = 'history/blinkpdf.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
try:
# Getting private key
container_env = subprocess.run(
["docker", "exec", "blinkpdf_container", "cat", "/opt/.env"],
capture_output=True,
text=True
).stdout.strip()
private_key = re.search(r'PRIVATE_KEY="(.+?)"', container_env).group(1)
assert len(private_key) > 0, 'Missing PRIVATE_KEY on .env'
sess = requests.Session()
# Checking C1: Login as user
url = f'http://localhost:{self.port}/login'
data = {"username": "user", "password": "user"}
r = sess.post(url, data=data, timeout=5)
assert 'Welcome to the PDF Signature App'.lower() in r.text.lower(), 'Cannot login as user'
pdfpath = 'files/blinkpdf_hellodocs.pdf'
pdfbytes = open(pdfpath, 'rb').read()
# Checking C2: Sign pdf as user
sign_url = f'http://localhost:{self.port}/sign'
r = sess.post(sign_url, timeout=5)
sendata = ('main.pdf', pdfbytes, 'application/pdf')
filedata = {'file': sendata}
r = sess.post(sign_url, files=filedata, timeout=5)
signed_pdf = r.content
assert r.status_code == 200 and 'application/pdf' in r.headers['Content-Type'], 'Signed PDF not available or incorrect content type'
signed_pdf_stream = io.BytesIO(signed_pdf)
assert verify_signature(signed_pdf_stream, private_key), 'Algorithm for signature process is changed'
# Checking C3: Verify valid pdf as user
pdf_bytes_stream = io.BytesIO(pdfbytes)
signed_pdf_stream = sign_pdf(pdf_bytes_stream, private_key)
verify_url = f'http://localhost:{self.port}/verify'
sendata = ('main_signed.pdf', signed_pdf_stream, 'application/pdf')
filedata = {'file': sendata}
r = sess.post(verify_url, files=filedata, timeout=5)
assert 'The signature is <strong>valid</strong>.' in r.text, 'Verify function not working or algoritm verify process is changed'
# Checking C3: Verify invalid pdf as user
verify_url = f'http://localhost:{self.port}/verify'
sendata = ('main_signed.pdf', pdfbytes, 'application/pdf')
filedata = {'file': sendata}
r = sess.post(verify_url, files=filedata, timeout=5)
assert 'The signature is <strong>invalid' in r.text, 'Verify function not working or algoritm verify process is changed for invalid signature'
# Checking C4: Checking flag on container
with open(self.flag_location, 'r') as f:
host_flag = f.read().strip()
container_flag = subprocess.run(
["docker", "exec", "blinkpdf_container", "cat", "/flag.txt"],
capture_output=True,
text=True
).stdout.strip()
assert host_flag == container_flag, 'Flag mismatch between host and container'
# Checking C5: Login as admin and enc_flag checking
url = f'http://localhost:{self.port}/login'
data = {'username': "admin", "password": f'{private_key}'}
r = sess.post(url, data=data, timeout=5)
assert 'Welcome to the PDF Signature App'.lower() in r.text.lower(), 'Cannot login as admin'
url = f'http://localhost:{self.port}/admin_panel'
r = sess.get(url, timeout=5)
enc_flag = r.text.split('encrypted flag: ')[1].split('</p>')[0]
cek, dec = decryptMessage(enc_flag, private_key)
assert dec.decode() == host_flag, 'Change algorithm for encryption flag'
assert cek == True, 'Change signature algorithm for encryption flag'
self.logger.info('Check passed for blinkpdf')
return True
except Exception as e:
self.logger.error(f'Could not check blinkpdf: {e}')
return False
receiver/challenges/Niko.py
-93
View File
@@ -1,93 +0,0 @@
from .Challenge import Challenge
import io
import requests
import random
import subprocess
import json
class Niko(Challenge):
flag_location = 'flags/niko.txt'
history_location = 'history/niko.txt'
def distribute(self, flag):
try:
with open(self.flag_location, 'w') as f:
f.write(flag)
with open(self.history_location, 'a') as f:
f.write(flag + '\n')
self.logger.info(f'Flag {flag} written to {self.flag_location}')
return True
except Exception as e:
self.logger.error(f'Could not write flag to {self.flag_location}: {e}')
return False
def check(self):
error_messages = [
"あなたはどんなオタクですか",
"冗談じゃないよ！",
"ｔｃｈ なんだよ こいつ",
"どうしてそんなことが可能でしょうか…不可能です",
"本当のあなたは何ですか？",
"うわー、ごめんなさい",
"御心のままに、主よ",
"もういいよ、やめて！",
"時間です",
"悪くないよ。"
]
try:
# Step 1: Check if the flag still exists and matches the one in the container
with open(self.flag_location, 'r') as f:
host_flag = f.read().strip()
container_flag = subprocess.run(
["docker", "exec", "niko_container", "cat", "/flag.txt"],
capture_output=True,
text=True
).stdout.strip()
assert host_flag == container_flag, 'Flag mismatch between host and container'
self.logger.info('Flag check passed for niko')
# Step 2: Check if can access flag
container_key = subprocess.run(
["docker", "exec", "niko_container", "cat", "/opt/flag"],
capture_output=True,
text=True
).stdout.strip()
urlFlag = f'http://localhost:{self.port}/api/getFlag?flag={container_key}'
expected_output = container_flag
response2 = requests.get(urlFlag)
assert response2.text.strip() == expected_output, 'Get Flag check does not work'
self.logger.info('Get flag endpoint check passed for niko')
# Step 3: Check if the webpage can be accessed
url = f'http://localhost:{self.port}/'
response = requests.get(url)
status_code = response.status_code
assert status_code == 200, 'Webpage is not accessible'
self.logger.info('Webpage accessibility check passed for niko')
# Step 4: Check if the output of the specific URL equals the expected string
urlFlag = f'http://localhost:{self.port}/api/getFlag?flag=1'
expected_output = "(⋟﹏⋞) 私をバカにしようとしているのか (´ ͡༎ຶ ͜ʖ ͡༎ຶ `)︵‿︵"
response2 = requests.get(urlFlag)
assert response2.text.strip() == expected_output, 'Webpage output does not match expected output'
self.logger.info('Webpage output check passed for niko')
# Step 5: Check if the chat endpoint is working
urlChat = f'http://localhost:{self.port}/api/chat'
data = 'test'
response3 = requests.post(urlChat, data=data)
assert json.loads(response3.text.strip()).get("output") in error_messages, 'Api endpoint is not accessible'
self.logger.info('Webpage delay check passed for niko')
return True
except Exception as e:
self.logger.error(f'Could not check niko: {e}')
return False
receiver/files/blinkpdf_hellodocs.pdf
BIN
View File
Binary file not shown.
receiver/flags.old/.gitkeep
View File
receiver/flags.old/art.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/back-to-basic.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/burvesigner.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/crawlback.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/gemas-fetcher.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/hirnfick.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/pasta.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/s3.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags.old/xl.txt
-1
View File
@@ -1 +0,0 @@
WRECKIT50{PLACEHOLDER}
receiver/flags/blinkpdf.txt
-1
View File
@@ -1 +0,0 @@
WreckIT50{PLACEHOLDER}
receiver/flags/blogpost.txt
+1
View File
@@ -0,0 +1 @@
GEMASTIK{PLACEHOLDER}
receiver/flags/cdn.txt
+1
View File
@@ -0,0 +1 @@
GEMASTIK{PLACEHOLDER}
receiver/flags/naraka.txt
-1
View File
@@ -1 +0,0 @@
WreckIT50{PLACEHOLDER}
receiver/flags/niko.txt
-1
View File
@@ -1 +0,0 @@
WreckIT50{PLACEHOLDER}
receiver/flags/poke.txt
-1
View File
@@ -1 +0,0 @@
WreckIT50{PLACEHOLDER}
receiver/flags/wanderer.txt
-1
View File
@@ -1 +0,0 @@
WreckIT50{PLACEHOLDER}
receiver/main.py
+2 -8
View File
@@ -4,9 +4,6 @@ from fastapi.security import HTTPBasic, HTTPBasicCredentials
from config import get_settings
from challenges.Poke import Poke
from challenges.Wanderer import Wanderer
from challenges.Naraka import Naraka
from challenges.Niko import Niko
from challenges.Blinkpdf import BlinkPDF
import os
@@ -16,11 +13,8 @@ security = HTTPBasic()
settings = get_settings()
challenges = {
"poke": Poke(10000),
"blinkpdf": BlinkPDF(11000),
"naraka": Naraka(12000),
"wanderer": Wanderer(13000),
"niko": Niko(15000),
"blogpost": Poke(10000),
"cdn": BlinkPDF(11000),
}
class Flag(BaseModel):
receiver/main.py.old
-116
View File
@@ -1,116 +0,0 @@
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from config import get_settings
from challenges.Art import Art
from challenges.XL import XL
from challenges.GemasNotes import GemasNotes
from challenges.Pasta import Pasta
from challenges.Burvesigner import Burvesigner
from challenges.S3 import S3
from challenges.Crawlback import Crawlback
from challenges.BackToBasic import BackToBasic
from challenges.GemasFetcher import GemasFetcher
from challenges.Hirnfick import Hirnfick
import os
app = FastAPI()
security = HTTPBasic()
settings = get_settings()
challenges = {
"art": Art(10000),
"xl": XL(11000),
"gemas-notes": GemasNotes(12000),
"pasta": Pasta(13000),
"burvesigner": Burvesigner(14000),
"hirnfick": Hirnfick(15000),
"gemas-fetcher": GemasFetcher(16000),
"s3": S3(20000),
"crawlback": Crawlback(21000),
"back-to-basic": BackToBasic(22000),
}
class Flag(BaseModel):
flag: str
challenge: str
class History(BaseModel):
log: str
@app.get("/")
def read_root():
return {"service": "receiver-service"}
@app.get("/restart/{challenge}")
def restart(challenge: str, credentials: HTTPBasicCredentials = Depends(security)):
validate(credentials, challenge)
os.system(f"docker compose -f {settings.COMPOSE_LOCATION} restart {challenge}")
return {"message": "Challenge restarted"}
@app.get("/rollback/{challenge}")
def rollback(challenge: str, credentials: HTTPBasicCredentials = Depends(security)):
validate(credentials, challenge)
os.system(f"docker compose -f {settings.COMPOSE_LOCATION} up -d --force-recreate {challenge}")
return {"message": "Challenge restarted"}
@app.get("/activate/{challenge}")
def activate(challenge: str, credentials: HTTPBasicCredentials = Depends(security)):
validate(credentials, challenge)
os.system(f"docker compose -f {settings.COMPOSE_LOCATION} up -d {challenge}")
return {"message": "Challenge activated"}
@app.get("/deactivate/{challenge}")
def deactive(challenge: str, credentials: HTTPBasicCredentials = Depends(security)):
validate(credentials, challenge)
os.system(f"docker compose -f {settings.COMPOSE_LOCATION} down {challenge}")
return {"message": "Challenge deactivated"}
@app.get("/credential/{challenge}")
def credential(challenge: str, credentials: HTTPBasicCredentials = Depends(security)):
validate(credentials, challenge)
return challenges[challenge].credentials()
@app.post("/flag")
def receive(data: Flag, credentials: HTTPBasicCredentials = Depends(security)):
validate(credentials, data.challenge)
challenge = challenges[data.challenge]
if challenge.distribute(data.flag):
return {"message": "Flag received"}
raise HTTPException(status_code=500, detail="Error receiving flag")
@app.get("/check/{challenge}")
def check(challenge: str, credentials: HTTPBasicCredentials = Depends(security)):
validate(credentials, challenge)
return {"success": challenges[challenge].check()}
@app.post("/history")
def history(data: History):
with open('history/command.txt', 'a') as f:
f.write(data.log + '\n')
return {"message": "Command received"}
def is_admin(credentials):
if credentials.username != settings.ADMIN_USERNAME or credentials.password != settings.ADMIN_PASSWORD:
return False
return True
def validate(credentials, challenge):
if not is_admin(credentials):
raise HTTPException(status_code=401, detail="Invalid credentials")
if challenge not in challenges:
raise HTTPException(status_code=400, detail="Invalid challenge")
receiver/modules/blinkpdf/__init__.py
-2
View File
@@ -1,2 +0,0 @@
from .signature import *
from .cipher import *
receiver/modules/blinkpdf/cipher.py
-43
View File
@@ -1,43 +0,0 @@
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Util.number import long_to_bytes, bytes_to_long
import hashlib
import random
import os
# This will be the base dont change it
q = 135589091449528481388008471290289910812753186702167314685052586130282290721619
p = 271178182899056962776016942580579821625506373404334629370105172260564581443239
g = pow(2, (p-1)//p, p)
def encryptMessage(message, PRIVATE_KEY):
key = hashlib.sha256(bytes.fromhex(PRIVATE_KEY)).digest()[:16]
cipher = AES.new(key, AES.MODE_CBC, iv=os.urandom(16))
ciphertext = cipher.iv + cipher.encrypt(pad(message,16))
digest_message = int(hashlib.sha256(message).hexdigest(), 16)
x = int(PRIVATE_KEY, 16)
rand = random.Random()
rand.seed(bytes_to_long(message))
k = rand.getrandbits(216)
r = pow(g, k, p) % q
s = (pow(k, -1, q) * (digest_message + r * x)) % q
y = pow(g, x, q)
signature = long_to_bytes(y).zfill(32).hex() + long_to_bytes(digest_message).zfill(32).hex() + long_to_bytes(r).zfill(32).hex() + long_to_bytes(s).zfill(32).hex()
return ciphertext.hex() + signature
def decryptMessage(ciphertext, PRIVATE_KEY):
cps = bytes.fromhex(ciphertext)
y = bytes_to_long(cps[-128:-96].lstrip(b'0'))
dig = bytes_to_long(cps[-96:-64].lstrip(b'0'))
r = bytes_to_long(cps[-64:-32].lstrip(b'0'))
s = bytes_to_long(cps[-32:].lstrip(b'0'))
u = pow(s, -1, q)
v = pow(g, (dig * u) % q, p) * pow(y, (r * u)%q, p) % p % q
ciphertext = cps[:-128]
iv = ciphertext[:16]
ct = ciphertext[16:]
key = hashlib.sha256(bytes.fromhex(PRIVATE_KEY)).digest()[:16]
cipher = AES.new(key, AES.MODE_CBC, iv=iv)
plaintext = cipher.decrypt(ct)
plain = unpad(plaintext, 16)
return v == r, plain
receiver/modules/blinkpdf/decdsa.py
-104
View File
@@ -1,104 +0,0 @@
import hashlib
import random
from ecdsa import NIST256p, ellipticcurve
class DECDSA:
def __init__(self, privateKey):
self.curve = NIST256p
self.order = self.curve.order
self.generator = self.curve.generator
self.private_key = int(privateKey, 16) % self.order
self.public_key = self.private_key * self.generator
# self.generate_keypair()
# def generate_keypair(self):
# test
# self.private_key = 68643326375728294502573326707893599968874260096336631364679496614035223206444
# self.private_key = random.randint(1, self.order - 1)
# self.public_key = self.private_key * self.generator
def lift_x(self, x):
p = self.curve.curve._CurveFp__p
a = self.curve.curve._CurveFp__a
b = self.curve.curve._CurveFp__b
y_squared = (x**3 + a*x + b) % p
y = pow(y_squared, (p + 1) // 4, p)
if (y * y) % p != y_squared:
raise ValueError(f"No valid point found for x={x}")
point1 = ellipticcurve.Point(self.curve.curve, x, y)
point2 = ellipticcurve.Point(self.curve.curve, x, p - y)
if y > p - y:
return point2
else:
return point1
def sign(self, message):
m1, m2 = message[:len(message)//2], message[len(message)//2:]
h1 = hashlib.sha256(m1).digest()[1:]
h2 = hashlib.sha256(m2).digest()[1:]
z1 = int.from_bytes(h1, byteorder='big') % self.order
z2 = int.from_bytes(h2, byteorder='big') % self.order
while True:
k1 = random.randint(z1, z1*4)
k2 = random.randint(z2, z2*4)
R1 = k1 * self.generator
R2 = k2 * self.generator
r1 = R1.x() % self.order
r2 = R2.x() % self.order
R_att_x = (self.lift_x(r1) + self.lift_x(r2)).x() % self.order
# assert for checking valid points
if(R_att_x!=(R1+R2).x() % self.order):
continue
if r1 == 0 or r2 == 0:
continue
ks = pow(k1, -1, self.order) + pow(k2, -1, self.order)
s = (pow(k1*k2, -1, self.order) * (z1 + r1 * self.private_key + z2 + r2 * self.private_key) * pow(ks, -1, self.order)) % self.order
if s == 0:
continue
r1, r2, s = int(r1), int(r2), int(s)
return self.sign_to_bytes(r1, r2, s)
def verify(self, message, signature):
r1, r2, s = self.bytes_to_sign(signature)
if not (1 <= r1 < self.order and 1 <= r2 < self.order and 1 <= s < self.order):
return False
m1, m2 = message[:len(message)//2], message[len(message)//2:]
h1 = hashlib.sha256(m1).digest()[1:]
h2 = hashlib.sha256(m2).digest()[1:]
z1 = int.from_bytes(h1, byteorder='big') % self.order
z2 = int.from_bytes(h2, byteorder='big') % self.order
s_inv = pow(s, -1, self.order)
u1 = (z1 * s_inv) % self.order
u2 = (z2 * s_inv) % self.order
u3 = (r1 * s_inv) % self.order
u4 = (r2 * s_inv) % self.order
R = u1 * self.generator + u3 * self.public_key + u2 * self.generator + u4 * self.public_key
R_x = R.x() % self.order
R_att_x = (self.lift_x(r1) + self.lift_x(r2)).x() % self.order
return R_x == R_att_x
def long_to_bytes(self, x):
return x.to_bytes(32, "big")
def bytes_to_long(self, x):
return int.from_bytes(x, "big")
def sign_to_bytes(self, r1, r2, s):
first_part = self.long_to_bytes(r1)
second_part = self.long_to_bytes(r2)
third_part = self.long_to_bytes(s)
return first_part + second_part + third_part
def bytes_to_sign(self, x):
r1 = self.bytes_to_long(x[:32])
r2 = self.bytes_to_long(x[32:64])
s = self.bytes_to_long(x[64:])
return r1, r2, s
receiver/modules/blinkpdf/signature.py
-43
View File
@@ -1,43 +0,0 @@
from .decdsa import DECDSA
import PyPDF2
import io
def sign_pdf(file, PRIVATE_KEY):
try:
decdsa = DECDSA(privateKey=PRIVATE_KEY)
pdf_reader = PyPDF2.PdfReader(file)
pdf_writer = PyPDF2.PdfWriter()
for page in pdf_reader.pages:
pdf_writer.add_page(page)
pdf_data = io.BytesIO()
pdf_writer.write(pdf_data)
pdf_data.seek(0)
pdf_content = pdf_data.read()
signature = decdsa.sign(pdf_content)
pdf_writer.add_metadata({'/Signature': signature.hex()})
signed_pdf = io.BytesIO()
pdf_writer.write(signed_pdf)
signed_pdf.seek(0)
return signed_pdf
except:
return False
def verify_signature(file, PRIVATE_KEY):
try:
decdsa = DECDSA(privateKey=PRIVATE_KEY)
pdf_reader = PyPDF2.PdfReader(file)
signature_text = pdf_reader.metadata.get('/Signature', '')
signature = bytes.fromhex(signature_text)
pdf_data = io.BytesIO()
pdf_writer = PyPDF2.PdfWriter()
for page in pdf_reader.pages:
pdf_writer.add_page(page)
pdf_writer.write(pdf_data)
pdf_data.seek(0)
pdf_content = pdf_data.read()
return decdsa.verify(pdf_content,signature)
except Exception as e:
print(e)
return False