{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "e5181b56-34a6-4c2e-8045-137f8190ad56",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "[*] Starting white-box phase...\n",
      "[*] Downloading model...\n",
      "[+] Model loaded: 89972 features\n",
      "  Attacking review wb_0... Done (6 words)\n",
      "  Attacking review wb_1... Done (3 words)\n",
      "  Attacking review wb_2... Done (3 words)\n",
      "  Attacking review wb_3... Done (4 words)\n",
      "  Attacking review wb_4... Done (6 words)\n",
      "  Attacking review wb_5... Done (11 words)\n",
      "  Attacking review wb_6... Done (5 words)\n",
      "  Attacking review wb_7... Done (4 words)\n",
      "  Attacking review wb_8... Done (24 words)\n",
      "  Attacking review wb_9... Done (7 words)\n",
      "[+] White-box phase: 10/10 completed\n",
      "[+] White-box phase completed successfully!\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import requests\n",
    "import pickle\n",
    "import numpy as np\n",
    "from typing import List, Dict, Tuple\n",
    "\n",
    "SEED = 1337\n",
    "np.random.seed(SEED)\n",
    "\n",
    "BASE_URL = \"http://154.57.164.67:30687\"\n",
    "\n",
    "class WhiteBoxAttacker:\n",
    "    def __init__(self, base_url: str):\n",
    "        self.base_url = base_url\n",
    "        self.model = None\n",
    "        self.vectorizer = None\n",
    "        self.feature_names = None\n",
    "        self.classes = None\n",
    "\n",
    "    def download_model(self):\n",
    "        print(\"[*] Downloading model...\")\n",
    "        r = requests.get(f\"{self.base_url}/model/download\")\n",
    "        r.raise_for_status()\n",
    "\n",
    "        with open(\"/tmp/model.pkl\", \"wb\") as f:\n",
    "            f.write(r.content)\n",
    "\n",
    "        with open(\"/tmp/model.pkl\", \"rb\") as f:\n",
    "            bundle = pickle.load(f)\n",
    "\n",
    "        self.model = bundle[\"classifier\"]\n",
    "        self.vectorizer = bundle[\"vectorizer\"]\n",
    "        self.feature_names = bundle[\"feature_names\"]\n",
    "        self.classes = bundle[\"classes\"]\n",
    "        print(f\"[+] Model loaded: {len(self.feature_names)} features\")\n",
    "\n",
    "    def calculate_word_scores(self, target_class: str) -> List[Tuple[str, float]]:\n",
    "        target_idx = self.classes.index(target_class)\n",
    "        other_idx = 1 - target_idx\n",
    "\n",
    "        scores = []\n",
    "        for i, feature in enumerate(self.feature_names):\n",
    "            if \" \" in feature:\n",
    "                continue\n",
    "            target_prob = np.exp(self.model.feature_log_prob_[target_idx][i])\n",
    "            other_prob = np.exp(self.model.feature_log_prob_[other_idx][i])\n",
    "            score = target_prob / (other_prob + 1e-10)\n",
    "            scores.append((feature, score))\n",
    "\n",
    "        scores.sort(key=lambda x: x[1], reverse=True)\n",
    "        return scores\n",
    "\n",
    "    def attack_review(self, text: str, target: str, max_words: int) -> Tuple[str, int]:\n",
    "        word_scores = self.calculate_word_scores(target)\n",
    "\n",
    "        augmented = text\n",
    "        for num_words in range(1, max_words + 1):\n",
    "            words_to_add = [w for w, _ in word_scores[:num_words]]\n",
    "            augmented = text + \" \" + \" \".join(words_to_add)\n",
    "\n",
    "            vec = self.vectorizer.transform([augmented])\n",
    "            prediction = self.model.predict(vec)[0]\n",
    "\n",
    "            if prediction == target:\n",
    "                return augmented, num_words\n",
    "\n",
    "        return augmented, max_words\n",
    "\n",
    "    def solve_whitebox(self) -> Dict:\n",
    "        print(\"\\n[*] Starting white-box phase...\")\n",
    "\n",
    "        r = requests.get(f\"{self.base_url}/challenge/whitebox\")\n",
    "        r.raise_for_status()\n",
    "        challenge = r.json()\n",
    "\n",
    "        reviews = challenge[\"reviews\"]\n",
    "        max_words = challenge[\"max_added_words\"]\n",
    "\n",
    "        self.download_model()\n",
    "\n",
    "        solutions = []\n",
    "        for review in reviews:\n",
    "            print(f\"  Attacking review {review['id']}...\", end=\" \")\n",
    "            augmented, words_used = self.attack_review(\n",
    "                review[\"text\"], review[\"target_sentiment\"], max_words\n",
    "            )\n",
    "            solutions.append({\"id\": review[\"id\"], \"augmented_text\": augmented})\n",
    "            print(f\"Done ({words_used} words)\")\n",
    "\n",
    "        r = requests.post(\n",
    "            f\"{self.base_url}/submit/whitebox\", json={\"solutions\": solutions}\n",
    "        )\n",
    "        r.raise_for_status()\n",
    "        result = r.json()\n",
    "\n",
    "        if \"results\" in result:\n",
    "            successes = sum(1 for r in result[\"results\"] if r.get(\"success\", False))\n",
    "            print(f\"[+] White-box phase: {successes}/10 completed\")\n",
    "\n",
    "        return result\n",
    "\n",
    "def main():\n",
    "    wb_attacker = WhiteBoxAttacker(BASE_URL)\n",
    "    wb_result = wb_attacker.solve_whitebox()\n",
    "\n",
    "    if not wb_result.get(\"phase_complete\", False):\n",
    "        print(\"[-] Failed to complete white-box phase\")\n",
    "        return\n",
    "\n",
    "    print(\"[+] White-box phase completed successfully!\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "ca580733-0f7d-401b-bd10-a29586b9f4b9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "[*] Starting black-box phase...\n",
      "  Attacking review bb_0... Done (22 words, 73 queries total)\n",
      "  Attacking review bb_1... Done (40 words, 164 queries total)\n",
      "  Attacking review bb_2... Done (20 words, 235 queries total)\n",
      "  Attacking review bb_3... Done (17 words, 303 queries total)\n",
      "  Attacking review bb_4... Done (4 words, 358 queries total)\n",
      "  Attacking review bb_5... Done (38 words, 447 queries total)\n",
      "  Attacking review bb_6... Done (40 words, 538 queries total)\n",
      "  Attacking review bb_7... Done (4 words, 593 queries total)\n",
      "  Attacking review bb_8... Done (13 words, 657 queries total)\n",
      "  Attacking review bb_9... Done (6 words, 714 queries total)\n",
      "[+] Black-box phase: 10/10 completed\n",
      "\n",
      "============================================================\n",
      "[+] SUCCESS! All challenges completed!\n",
      "[+] Flag: HTB{f34tur3_0bfu5c4t10n_m45t3r3d}\n",
      "============================================================\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import requests\n",
    "import numpy as np\n",
    "from typing import List, Dict, Tuple\n",
    "\n",
    "SEED = 1337\n",
    "np.random.seed(SEED)\n",
    "\n",
    "BASE_URL = \"http://154.57.164.67:30687\"\n",
    "\n",
    "class BlackBoxAttacker:\n",
    "    def __init__(self, base_url: str):\n",
    "        self.base_url = base_url\n",
    "        self.word_impacts = {}\n",
    "        self.query_count = 0\n",
    "\n",
    "    def predict(self, text: str) -> Dict:\n",
    "        self.query_count += 1\n",
    "        r = requests.post(f\"{self.base_url}/predict\", json={\"text\": text})\n",
    "        r.raise_for_status()\n",
    "        return r.json()\n",
    "\n",
    "    def estimate_word_impacts(\n",
    "        self, base_text: str, vocabulary: List[str]\n",
    "    ) -> List[Tuple[str, float]]:\n",
    "        base_pred = self.predict(base_text)\n",
    "        base_pos_prob = base_pred[\"positive_probability\"]\n",
    "\n",
    "        impacts = []\n",
    "        for word in vocabulary:\n",
    "            augmented = base_text + \" \" + word\n",
    "            pred = self.predict(augmented)\n",
    "            impact = pred[\"positive_probability\"] - base_pos_prob\n",
    "            impacts.append((word, impact))\n",
    "\n",
    "        impacts.sort(key=lambda x: x[1], reverse=True)\n",
    "        return impacts\n",
    "\n",
    "    def attack_review(self, text: str, max_words: int) -> Tuple[str, int]:\n",
    "        positive_vocabulary = [\n",
    "            \"excellent\", \"amazing\", \"wonderful\", \"fantastic\", \"brilliant\",\n",
    "            \"outstanding\", \"superb\", \"magnificent\", \"perfect\", \"exceptional\",\n",
    "            \"masterpiece\", \"genius\", \"beautiful\", \"stunning\", \"remarkable\",\n",
    "            \"awesome\", \"incredible\", \"phenomenal\", \"spectacular\", \"marvelous\",\n",
    "            \"great\", \"good\", \"love\", \"loved\", \"best\", \"favorite\", \"enjoyed\",\n",
    "            \"recommend\", \"highly\", \"definitely\", \"must\", \"liked\", \"appreciate\",\n",
    "            \"admire\", \"adore\", \"enjoy\", \"compelling\", \"engaging\", \"captivating\",\n",
    "            \"mesmerizing\", \"powerful\", \"touching\", \"moving\", \"inspiring\",\n",
    "            \"uplifting\", \"heartwarming\", \"clever\", \"witty\", \"funny\", \"hilarious\",\n",
    "            \"entertaining\"\n",
    "        ]\n",
    "\n",
    "        impacts = self.estimate_word_impacts(text, positive_vocabulary[:50])\n",
    "\n",
    "        augmented = text\n",
    "        for i, (word, impact) in enumerate(impacts, 1):\n",
    "            if i > max_words:\n",
    "                break\n",
    "\n",
    "            augmented = augmented + \" \" + word\n",
    "            pred = self.predict(augmented)\n",
    "\n",
    "            if pred[\"label\"] == \"positive\":\n",
    "                return augmented, i\n",
    "\n",
    "        if impacts:\n",
    "            top_words = [w for w, _ in impacts[:10]]\n",
    "            words_to_add = []\n",
    "            while len(words_to_add) < max_words:\n",
    "                words_to_add.extend(top_words)\n",
    "            augmented = text + \" \" + \" \".join(words_to_add[:max_words])\n",
    "\n",
    "        return augmented, max_words\n",
    "\n",
    "    def solve_blackbox(self) -> Dict:\n",
    "        print(\"\\n[*] Starting black-box phase...\")\n",
    "\n",
    "        r = requests.get(f\"{self.base_url}/challenge/blackbox\")\n",
    "        r.raise_for_status()\n",
    "        challenge = r.json()\n",
    "\n",
    "        reviews = challenge[\"reviews\"]\n",
    "        max_words = challenge[\"max_added_words\"]\n",
    "\n",
    "        solutions = []\n",
    "        for review in reviews:\n",
    "            print(f\"  Attacking review {review['id']}...\", end=\" \")\n",
    "            augmented, words_used = self.attack_review(review[\"text\"], max_words)\n",
    "            solutions.append({\"id\": review[\"id\"], \"augmented_text\": augmented})\n",
    "            print(f\"Done ({words_used} words, {self.query_count} queries total)\")\n",
    "\n",
    "        r = requests.post(\n",
    "            f\"{self.base_url}/submit/blackbox\", json={\"solutions\": solutions}\n",
    "        )\n",
    "        r.raise_for_status()\n",
    "        result = r.json()\n",
    "\n",
    "        if \"results\" in result:\n",
    "            successes = sum(1 for r in result[\"results\"] if r.get(\"success\", False))\n",
    "            print(f\"[+] Black-box phase: {successes}/10 completed\")\n",
    "\n",
    "        return result\n",
    "\n",
    "def main():\n",
    "    bb_attacker = BlackBoxAttacker(BASE_URL)\n",
    "    bb_result = bb_attacker.solve_blackbox()\n",
    "\n",
    "    if bb_result.get(\"flag\"):\n",
    "        print(\"\\n\" + \"=\" * 60)\n",
    "        print(\"[+] SUCCESS! All challenges completed!\")\n",
    "        print(f\"[+] Flag: {bb_result['flag']}\")\n",
    "        print(\"=\" * 60)\n",
    "    else:\n",
    "        print(\"[-] Failed to complete black-box phase\")\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0b9ce2a6-d30d-466c-9e79-e4054b13b4ce",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}