#!/usr/bin/env python3 import sys import json import time import random import requests from typing import Optional from checklib import * from jform_lib import CheckMachine class Checker(BaseChecker): vulns: int = 1 timeout: int = 15 uses_attack_data: bool = True def __init__(self, host: str): super().__init__(host) self.m = CheckMachine(self) def _new_session(self) -> requests.Session: return get_initialized_session() def _require_json(self, r: requests.Response, where: str) -> dict: self.assert_in("application/json", r.headers.get("Content-Type", ""), f"{where}: Content-Type is not JSON") return self.get_json(r, f"{where}: malformed JSON") def _extract_form_id(self, obj: dict) -> str: form_id = obj.get("id") or obj.get("formId") or obj.get("_id") self.assert_neq(None, form_id, "Service returned empty form id") form_id = str(form_id) self.assert_neq("", form_id, "Service returned empty form id") return form_id def _register_and_login(self, s: requests.Session): username = rnd_username() password = rnd_password() r = self.m.register(s, username, password) self.assert_eq(200, r.status_code, "Service /signup returned unexpected status code") _ = self._require_json(r, "/signup") r = self.m.login(s, username, password) self.assert_eq(200, r.status_code, "Service /login returned unexpected status code") _ = self._require_json(r, "/login") r = self.m.me(s) self.assert_eq(200, r.status_code, "Service /me returned unexpected status code") me = self._require_json(r, "/me") self.assert_in("username", me, "Service /me missing username") self.assert_eq(me["username"], username, "Service /me returned wrong username") self.assert_in("userId", me, "Service /me missing userId") user_id = str(me["userId"]) self.assert_neq("", user_id, "Service /me returned empty userId") return username, password, user_id def _form_catalog(self): return [ { "title": "Обратная связь", "description": "Короткий опрос о качестве сервиса", "fields": [ {"name": "email_1", "type": "email", "label": "Email", "placeholder": "you@example.ru", "required": True}, {"name": "number_1", "type": "number", "label": "Оценка (1-10)", "placeholder": "10", "required": True}, {"name": "textarea_1", "type": "textarea", "label": "Комментарий", "placeholder": "Напишите пару слов", "required": False}, ], }, { "title": "Заявка на работу", "description": "Расскажите о себе", "fields": [ {"name": "text_1", "type": "text", "label": "ФИО", "placeholder": "Иванов Иван", "required": True}, {"name": "email_1", "type": "email", "label": "Почта", "placeholder": "you@example.ru", "required": True}, {"name": "number_1", "type": "number", "label": "Опыт (лет)", "placeholder": "3", "required": True}, {"name": "textarea_1", "type": "textarea", "label": "Резюме (кратко)", "placeholder": "Ключевые навыки...", "required": False}, ], }, { "title": "Регистрация на событие", "description": "Подтвердите участие", "fields": [ {"name": "text_1", "type": "text", "label": "Имя", "placeholder": "Ваше имя", "required": True}, {"name": "email_1", "type": "email", "label": "Email", "placeholder": "you@example.ru", "required": True}, {"name": "number_1", "type": "number", "label": "Количество гостей", "placeholder": "0", "required": False}, {"name": "text_2", "type": "text", "label": "Дата", "placeholder": "2025-10-05", "required": True}, ], }, { "title": "Сообщить об ошибке", "description": "Помогите нам стать лучше", "fields": [ {"name": "text_1", "type": "text", "label": "Заголовок", "placeholder": "Коротко о проблеме", "required": True}, {"name": "select_1", "type": "select", "label": "Серьёзность", "options": ["Низкая", "Средняя", "Высокая"], "required": True}, {"name": "textarea_1", "type": "textarea", "label": "Шаги воспроизведения", "placeholder": "1) ... 2) ...", "required": True}, ], }, { "title": "Заказ пиццы", "description": "Соберите свою пиццу", "fields": [ {"name": "select_1", "type": "select", "label": "Размер", "options": ["Маленькая", "Средняя", "Большая"], "required": True}, {"name": "text_1", "type": "text", "label": "Адрес", "placeholder": "Город, улица, дом", "required": True}, {"name": "select_2", "type": "select", "label": "Соус", "options": ["Томатный", "Сливочный", "Барбекю"], "required": False}, ], }, { "title": "Опрос о цветах", "description": "Пара вопросов о вкусе", "fields": [ {"name": "number_1", "type": "number", "label": "Возраст", "placeholder": "25", "required": False}, {"name": "text_1", "type": "text", "label": "Любимый цвет", "placeholder": "синий", "required": True}, ], }, { "title": "Мини-викторина", "description": "Два коротких вопроса", "fields": [ {"name": "text_1", "type": "text", "label": "Имя", "placeholder": "Ваше имя", "required": True}, {"name": "number_1", "type": "number", "label": "Сколько будет 2+2?", "placeholder": "4", "required": True}, {"name": "number_2", "type": "number", "label": "Сколько дней в неделе?", "placeholder": "7", "required": True}, ], }, { "title": "Заявка в поддержку", "description": "Опишите проблему", "fields": [ {"name": "email_1", "type": "email", "label": "Email", "placeholder": "you@example.ru", "required": True}, {"name": "text_1", "type": "text", "label": "Тема", "placeholder": "Коротко", "required": True}, {"name": "textarea_1", "type": "textarea", "label": "Описание", "placeholder": "Что произошло?", "required": True}, ], }, { "title": "Командировка", "description": "Заявка на поездку", "fields": [ {"name": "text_1", "type": "text", "label": "Откуда", "placeholder": "Москва", "required": True}, {"name": "text_2", "type": "text", "label": "Куда", "placeholder": "Санкт-Петербург", "required": True}, {"name": "text_3", "type": "text", "label": "Дата выезда", "placeholder": "2025-10-05", "required": True}, {"name": "text_4", "type": "text", "label": "Дата возврата", "placeholder": "2025-10-07", "required": False}, {"name": "number_1", "type": "number", "label": "Бюджет, ₽", "placeholder": "15000", "required": False}, ], }, { "title": "Рецензия на книгу", "description": "Поделитесь впечатлениями", "fields": [ {"name": "text_1", "type": "text", "label": "Название", "placeholder": "Книга", "required": True}, {"name": "text_2", "type": "text", "label": "Автор", "placeholder": "Автор", "required": True}, {"name": "number_1", "type": "number", "label": "Оценка (1-10)", "placeholder": "8", "required": True}, {"name": "textarea_1", "type": "textarea", "label": "Отзыв", "placeholder": "Ваши мысли...", "required": False}, ], }, ] def _pick_schema(self, seed: int): rng = random.Random(seed) base = self._form_catalog()[rng.randrange(10)] schema = { "title": base["title"], "description": base.get("description", ""), "fields": [dict(f) for f in base["fields"]], } return schema, rng def _make_flag_sentence(self, flag: str, rng: random.Random) -> str: templates = [ "Сегодняшний код дня — {flag} — случайно появился в заметках.", "Если что, подсказка спрятана так: {flag}.", "Запомни шифр: {flag}, пригодится позже.", "Пока никто не видит, оставлю тут {flag}.", "Лозунг встречи: \"{flag}\" — без комментариев.", "Пароль на вход: {flag}; не перепутай!", "Метка для отчёта — {flag} — проверь внимательно.", "Секретный ингредиент рецепта: {flag}.", "В списке дел первым пунктом стоит {flag}.", "На полях тетради выведено: {flag}.", ] return rng.choice(templates).format(flag=flag) def _make_valid_payload(self, schema: dict, username: str, rng: random.Random, flag: Optional[str] = None) -> dict: def word(n=8): letters = "абвгдеёжзийклмнопрстуфхцчшщэюя" return "".join(rng.choice(letters) for _ in range(n)) email = f"{username}@example.ru" text_like = [f for f in schema.get("fields", []) if f.get("type") in ("text", "textarea")] flag_field_name = rng.choice(text_like)["name"] if (flag and text_like) else None sentence = self._make_flag_sentence(flag, rng) if flag else None payload = {} for f in schema.get("fields", []): t = f.get("type") name = f.get("name") req = bool(f.get("required")) if name == flag_field_name and sentence is not None: payload[name] = sentence continue if t == "email": payload[name] = email elif t == "number": payload[name] = rng.randint(1, 100) elif t == "textarea": payload[name] = f"{word(6)} {word(6)} {word(6)} {word(6)}" elif t == "select": options = f.get("options") or [] payload[name] = rng.choice(options) if options else "Вариант 1" else: payload[name] = word(8) if req else word(5) return payload def check(self): try: s1 = self._new_session() u1, p1, id1 = self._register_and_login(s1) s2 = self._new_session() u2, p2, id2 = self._register_and_login(s2) try: id1_i = int(id1); id2_i = int(id2) except Exception as e: self.cquit(Status.MUMBLE, "Service returned non-integer userId", str(e)) self.assert_eq(True, id2_i > id1_i, "Service produced incorrect userId ordering") schema1, rng = self._pick_schema(seed=id2_i) schema2, _ = self._pick_schema(seed=id2_i + 777) r = self.m.create_form(s2, schema1) self.assert_eq(200, r.status_code, "Service create_form returned unexpected status code") created1 = self._require_json(r, "create_form #1") form_id1 = self._extract_form_id(created1) self.assert_in("formNumber", created1, "Service did not return formNumber on creation") num1 = int(created1["formNumber"]) r = self.m.get_form(s2, form_id1) self.assert_eq(200, r.status_code, "Service get_form returned unexpected status code") gf1 = self._require_json(r, "get_form #1") self.assert_eq(form_id1, str(gf1.get("id") or gf1.get("formId") or gf1.get("_id") or ""), "Service get_form returned wrong form id") self.assert_eq(num1, int(gf1.get("formNumber")), "Service get_form returned wrong formNumber") r = self.m.create_form(s2, schema2) self.assert_eq(200, r.status_code, "Service create_form returned unexpected status code") created2 = self._require_json(r, "create_form #2") form_id2 = self._extract_form_id(created2) self.assert_in("formNumber", created2, "Service did not return formNumber on creation") num2 = int(created2["formNumber"]) self.assert_eq(True, num2 > num1, "Service produced non-increasing formNumber") self.assert_neq(form_id1, form_id2, "Service assigned duplicate form identifiers") r = self.m.get_my_forms(s2) self.assert_eq(200, r.status_code, "Service get_my_forms returned unexpected status code") my = self._require_json(r, "get_my_forms") self.assert_eq(True, isinstance(my, list), "Service returned invalid forms list") ids_in_list = {str(i.get("id") or i.get("formId") or i.get("_id") or "") for i in my if isinstance(i, dict)} self.assert_in(form_id1, ids_in_list, "Service did not list previously created form #1") self.assert_in(form_id2, ids_in_list, "Service did not list previously created form #2") payload = self._make_valid_payload(schema2, username=u2, rng=random.Random(id2_i)) r = self.m.submit_form_response(s2, form_id2, payload) self.assert_eq(200, r.status_code, "Service submit_form_response returned unexpected status code") self._require_json(r, "submit_form_response") r = self.m.get_form_results(s2, form_id2) self.assert_eq(200, r.status_code, "Service get_form_results returned unexpected status code") res_after = self._require_json(r, "get_form_results after submit") self.assert_eq(True, isinstance(res_after, list), "Service returned invalid results list") self.assert_eq(True, len(res_after) >= 1, "Service did not persist submitted response") self.cquit(Status.OK) except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: self.cquit(Status.DOWN, "Service is down or unreachable", str(e)) def put(self, flag_id: str, flag: str, vuln: str): try: s = self._new_session() username, password, user_id = self._register_and_login(s) user_id_int = int(user_id) schema, rng = self._pick_schema(seed=user_id_int) r = self.m.create_form(s, schema) self.assert_eq(200, r.status_code, "Service create_form returned unexpected status code") created = self._require_json(r, "create_form") form_id = self._extract_form_id(created) self.assert_in("formNumber", created, "Service did not return formNumber on creation") _ = int(created["formNumber"]) payload = self._make_valid_payload(schema, username=username, rng=rng, flag=flag) r = self.m.submit_form_response(s, form_id, payload) self.assert_eq(200, r.status_code, "Service submit_form_response returned unexpected status code") _ = self._require_json(r, "submit_form_response") public = json.dumps({"userId": user_id, "username": username}) private = json.dumps({"username": username, "password": password, "form_id": form_id}) self.cquit(Status.OK, public, private) except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: self.cquit(Status.DOWN, "Service is down or unreachable", str(e)) def get(self, flag_id: str, flag: str, vuln: str): try: s = self._new_session() try: blob = json.loads(flag_id) except Exception as e: self.cquit(Status.CORRUPT, "Invalid private data", str(e)) if isinstance(blob, dict) and {"username", "password", "form_id"} <= set(blob.keys()): priv = blob elif isinstance(blob, dict) and isinstance(blob.get("private"), dict): priv = blob["private"] else: self.cquit(Status.CORRUPT, "Missing auth and form_id fields", f"got: {sorted(blob.keys()) if isinstance(blob, dict) else type(blob)}") username = priv.get("username") password = priv.get("password") form_id = str(priv.get("form_id") or "") self.assert_eq(True, all([username, password, form_id]), "Missing required fields to access form") r = self.m.login(s, username, password) self.assert_eq(200, r.status_code, "Service /login returned unexpected status code") _ = self._require_json(r, "/login (GET)") r = self.m.get_form_results(s, form_id) self.assert_eq(200, r.status_code, "Service get_form_results returned unexpected status code") res = self._require_json(r, "get_form_results (GET)") self.assert_eq(True, isinstance(res, list), "Service returned invalid results list") def contains_flag(obj) -> bool: if isinstance(obj, str): if flag in obj: return True try: inner = json.loads(obj) except Exception: return False return contains_flag(inner) if isinstance(obj, dict): for k in ("response", "data", "payload", "answers"): if k in obj and contains_flag(obj[k]): return True for v in obj.values(): if contains_flag(v): return True return False if isinstance(obj, list): for v in obj: if contains_flag(v): return True return False return False found = any(contains_flag(item) for item in res) self.assert_eq(True, found, "Service does not return previously submitted data (flag not found)") self.cquit(Status.OK) except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e: self.cquit(Status.DOWN, "Service is down or unreachable", str(e)) if __name__ == '__main__': c = Checker(sys.argv[2]) try: c.action(sys.argv[1], *sys.argv[3:]) except c.get_check_finished_exception(): cquit(Status(c.status), c.public, c.private)