diff --git a/services/blogpost/sla.py b/services/blogpost/sla.py index ecf4f5c..43ffbd3 100644 --- a/services/blogpost/sla.py +++ b/services/blogpost/sla.py @@ -29,7 +29,7 @@ class WebAppSLA(Challenge): flag_location = 'flags/webapp.txt' # Host copy (used by your orchestrator) history_location = 'history/webapp.txt' container_flag_path = '/flag.txt' - container_name = 'chal_app' # <-- set to your actual container name + container_name = 'blogpost_container' # <-- set to your actual container name # Heuristics to recognize ExifTool output _exif_markers = ( diff --git a/services/cdn/sla.py b/services/cdn/sla.py new file mode 100644 index 0000000..9c22e9c --- /dev/null +++ b/services/cdn/sla.py @@ -0,0 +1,205 @@ +import io +import re +import random +import string +import logging +import subprocess +from typing import Optional + +import requests +from PIL import Image + +from .Challenge import Challenge + + +class WebAppSLA(Challenge): + """ + SLA checker for the Flask app. + + Enforced rules (HARD FAIL if violated): + 1) /flag.txt exists in container AND equals host copy at flags/webapp.txt + 2) Image metadata generation works via ExifTool and is retrievable from /post/?meta=1 + 3) exiftool is installed in the container and is the provider of that metadata + """ + + # Paths & infra knobs + flag_location = 'flags/webapp.txt' # host copy (written by your orchestrator) + history_location = 'history/webapp.txt' + container_flag_path = '/flag.txt' + container_name = 'chal_app' # adjust if your container name differs + + # Canonical ExifTool markers (NO app fallbacks allowed) + _exif_must_have = ( + 'File Name', + 'Date Created', + ) + + def _make_logger(self): + if not hasattr(self, 'logger') or self.logger is None: + self.logger = logging.getLogger(self.__class__.__name__) + if not self.logger.handlers: + h = logging.StreamHandler() + h.setFormatter(logging.Formatter('[%(levelname)s] %(message)s')) + self.logger.addHandler(h) + self.logger.setLevel(logging.INFO) + + # --- Helpers --- + def _gen_username(self) -> str: + return "user_" + ''.join(random.choices(string.ascii_lowercase + string.digits, k=8)) + + def _gen_password(self) -> str: + return "Pw!" + ''.join(random.choices(string.ascii_letters + string.digits, k=10)) + + def _tiny_png_bytes(self) -> bytes: + img = Image.new("RGB", (2, 2), (123, 200, 50)) + buf = io.BytesIO() + img.save(buf, format="PNG") + return buf.getvalue() + + def _docker_exec(self, args: list[str], timeout: int = 10) -> subprocess.CompletedProcess: + return subprocess.run( + ["docker", "exec", self.container_name] + args, + capture_output=True, text=True, timeout=timeout + ) + + # --- SLA core --- + def check(self) -> bool: + self._make_logger() + base_url = f"http://localhost:{self.port}" + s = requests.Session() + + # [1/9] Liveness + self.logger.info("[1/9] Checking liveness (/login) ...") + try: + r = s.get(base_url + "/login", timeout=10) + assert r.status_code == 200, f"/login HTTP {r.status_code}" + self.logger.info(" ✓ Login page reachable") + except Exception as e: + self.logger.error(f" ✗ Liveness failed: {e}") + return False + + # [2/9] exiftool MUST exist in container + self.logger.info("[2/9] Verifying exiftool installed in container ...") + try: + proc = self._docker_exec(["/bin/sh", "-lc", "command -v exiftool && exiftool -ver"]) + out = (proc.stdout or "").strip() + assert proc.returncode == 0 and out, "exiftool not found or not runnable" + self.logger.info(f" ✓ exiftool OK (version: {out.splitlines()[-1]})") + except Exception as e: + self.logger.error(f" ✗ exiftool check failed: {e}") + return False + + # [3/9] flag.txt MUST exist and MUST equal host copy + self.logger.info("[3/9] Verifying /flag.txt exists and matches host copy ...") + try: + # read container flag + proc = self._docker_exec(["/bin/sh", "-lc", f"test -f {self.container_flag_path} && cat {self.container_flag_path}"]) + assert proc.returncode == 0, "flag.txt missing in container" + container_flag = (proc.stdout or "").strip() + assert container_flag, "flag.txt empty" + + # read host copy + with open(self.flag_location, "r", encoding="utf-8") as fh: + host_flag = fh.read().strip() + + # exact match + assert container_flag == host_flag, "flag.txt content changed/mismatch" + self.logger.info(" ✓ flag.txt present and unchanged") + except Exception as e: + self.logger.error(f" ✗ Flag verification failed: {e}") + return False + + # [4/9] Register + self.logger.info("[4/9] Registering a fresh user ...") + username = self._gen_username() + password = self._gen_password() + try: + r = s.post(base_url + "/register", + data={"username": username, "password": password}, + allow_redirects=True, timeout=10) + assert 200 <= r.status_code < 400, f"Register HTTP {r.status_code}" + self.logger.info(f" ✓ Registered as {username}") + except Exception as e: + self.logger.error(f" ✗ Registration failed: {e}") + return False + + # [5/9] Login + self.logger.info("[5/9] Logging in ...") + try: + r = s.post(base_url + "/login", + data={"username": username, "password": password}, + allow_redirects=True, timeout=10) + assert r.status_code == 200, f"Login HTTP {r.status_code}" + assert "Invalid credentials" not in r.text, "Invalid credentials" + self.logger.info(" ✓ Logged in") + except Exception as e: + self.logger.error(f" ✗ Login failed: {e}") + return False + + # [6/9] Upload tiny PNG (triggers exiftool path in app) + self.logger.info("[6/9] Uploading PNG to generate metadata ...") + title = "SLA Post " + ''.join(random.choices(string.ascii_letters, k=6)) + files = {"image": ("probe.png", self._tiny_png_bytes(), "image/png")} + data = {"title": title} + try: + r = s.post(base_url + "/upload", files=files, data=data, + allow_redirects=True, timeout=20) + assert 200 <= r.status_code < 400, f"Upload HTTP {r.status_code}" + self.logger.info(" ✓ Upload OK") + except Exception as e: + self.logger.error(f" ✗ Upload failed: {e}") + return False + + # [7/9] Find newest post id (prefer near our title) + self.logger.info("[7/9] Resolving new post id from /gallery ...") + try: + r = s.get(base_url + "/gallery", timeout=10) + if r.status_code != 200: + r = s.get(base_url + "/", timeout=10) + assert r.status_code == 200, f"Index HTTP {r.status_code}" + + post_id: Optional[int] = None + for m in re.finditer(re.escape(title), r.text): + window = r.text[max(0, m.start()-1000): m.start()+1000] + mm = re.search(r'/post/(\d+)', window) + if mm: + post_id = int(mm.group(1)); break + if post_id is None: + ids = re.findall(r'/post/(\d+)', r.text) + assert ids, "No /post/ links found" + post_id = max(map(int, ids)) + + self.logger.info(f" ✓ Post id = {post_id}") + except Exception as e: + self.logger.error(f" ✗ Post discovery failed: {e}") + return False + + # [8/9] Post renders + self.logger.info("[8/9] Verifying post page renders ...") + try: + vp = s.get(base_url + f"/post/{post_id}", timeout=10) + assert vp.status_code == 200, f"Post HTTP {vp.status_code}" + assert title in vp.text, "Post title missing" + self.logger.info(" ✓ Post page OK") + except Exception as e: + self.logger.error(f" ✗ View post failed: {e}") + return False + + # [9/9] Metadata must be real ExifTool output (no fallbacks) + self.logger.info("[9/9] Verifying metadata via /post/ (ExifTool) ...") + try: + mr = s.get(base_url + f"/post/{post_id}", timeout=10) + assert mr.status_code == 200, f"Meta HTTP {mr.status_code}" + meta = (mr.text or "").strip() + assert meta, "Empty metadata" + # refuse app fallbacks + assert "no-metadata" not in meta and "exif_err:" not in meta, "App fallback metadata detected" + # must contain canonical ExifTool keys + assert all(k in meta for k in self._exif_must_have), "Metadata is not ExifTool output" + self.logger.info(" ✓ Metadata present and produced by ExifTool") + except Exception as e: + self.logger.error(f" ✗ Metadata verification failed: {e}") + return False + + self.logger.info("SLA check passed ✅") + return True