225 lines
9.2 KiB
Python
225 lines
9.2 KiB
Python
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 CDN(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/<id>?meta=1
|
|
3) exiftool is installed in the container and is the provider of that metadata
|
|
"""
|
|
|
|
# Paths & infra knobs
|
|
flag_location = 'flags/cdn.txt' # host copy (written by your orchestrator)
|
|
history_location = 'history/cdn.txt'
|
|
container_flag_path = '/flag.txt'
|
|
container_name = 'cdn_container' # adjust if your container name differs
|
|
|
|
# Canonical ExifTool markers (NO app fallbacks allowed)
|
|
_exif_must_have = (
|
|
'File Name',
|
|
'Date Created',
|
|
)
|
|
|
|
# --- Flag distribution hook (optional, mirrors your example) ---
|
|
def distribute(self, flag: str) -> bool:
|
|
"""
|
|
Writes/records the current flag on the host. Your infra may separately
|
|
mount/copy it into the container; this class *also* verifies existence
|
|
inside the container during .check().
|
|
"""
|
|
self._make_logger()
|
|
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"Failed writing host flag: {e}")
|
|
return False
|
|
|
|
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/<id> 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/<id> (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
|