Merge branch 'main' of https://github.com/rayhanhanaputra/gemastik18-final

receiver/challenges/CDN.py

@@ -111,17 +111,14 @@ class CDN(Challenge):
         # [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:
@@ -169,26 +166,50 @@ class CDN(Challenge):
             self.logger.error(f"    ✗ Upload failed: {e}")
             return False
 
-        # [7/9] Find newest post id (prefer near our title)
+        # [7/9] Resolve post id from the JUST-CREATED post (not the newest in gallery)
-        self.logger.info("[7/9] Resolving new post id from /gallery ...")
+        self.logger.info("[7/9] Resolving post id from upload result ...")
         try:
-            r = s.get(base_url + "/gallery", timeout=10)
+            created_post_id: Optional[int] = None
-            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
+            # Check redirect chain and final URL for /post/<id>
-            for m in re.finditer(re.escape(title), r.text):
+            candidate_urls = []
-                window = r.text[max(0, m.start()-1000): m.start()+1000]
+            if r.history:
-                mm = re.search(r'/post/(\d+)', window)
+                for resp in r.history:
+                    loc = resp.headers.get("Location", "")
+                    if loc:
+                        candidate_urls.append(loc if loc.startswith("http") else base_url + loc)
+            candidate_urls.append(getattr(r, "url", ""))
+
+            for u in candidate_urls:
+                m = re.search(r'/post/(\d+)', u or "")
+                if m:
+                    created_post_id = int(m.group(1))
+                    break
+
+            # If still None, try to parse the response body for a /post/<id> link
+            if created_post_id is None:
+                mm = re.search(r'/post/(\d+)', r.text or "")
                 if mm:
-                    post_id = int(mm.group(1)); break
+                    created_post_id = int(mm.group(1))
-            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}")
+            # LAST RESORT: fallback to gallery by matching our unique title window (kept for robustness)
+            if created_post_id is None:
+                self.logger.info("    ↪ Falling back to /gallery title match ...")
+                gr = s.get(base_url + "/gallery", timeout=10)
+                if gr.status_code != 200:
+                    gr = s.get(base_url + "/", timeout=10)
+                    assert gr.status_code == 200, f"Index HTTP {gr.status_code}"
+
+                post_id: Optional[int] = None
+                for m in re.finditer(re.escape(title), gr.text):
+                    window = gr.text[max(0, m.start()-1000): m.start()+1000]
+                    mm2 = re.search(r'/post/(\d+)', window)
+                    if mm2:
+                        post_id = int(mm2.group(1)); break
+                assert post_id is not None, "Could not resolve created post id"
+                created_post_id = post_id
+
+            self.logger.info(f"    ✓ Post id = {created_post_id}")
         except Exception as e:
             self.logger.error(f"    ✗ Post discovery failed: {e}")
             return False
@@ -196,7 +217,7 @@ class CDN(Challenge):
         # [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)
+            vp = s.get(base_url + f"/post/{created_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")
@@ -204,18 +225,22 @@ class CDN(Challenge):
             self.logger.error(f"    ✗ View post failed: {e}")
             return False
 
-        # [9/9] Metadata must be real ExifTool output (no fallbacks)
+        # [9/9] Metadata must appear somewhere in the post HTML (no fallback)
-        self.logger.info("[9/9] Verifying metadata via /post/<id> (ExifTool) ...")
+        self.logger.info("[9/9] Verifying metadata presence (File Name / Date Created) ...")
         try:
-            mr = s.get(base_url + f"/post/{post_id}", timeout=10)
+            mr = s.get(base_url + f"/post/{created_post_id}", timeout=10)
             assert mr.status_code == 200, f"Meta HTTP {mr.status_code}"
-            meta = (mr.text or "").strip()
+            html = (mr.text or "")
-            assert meta, "Empty metadata"
+            assert html.strip(), "Empty metadata HTML"
-            # refuse app fallbacks
+
-            assert "no-metadata" not in meta and "exif_err:" not in meta, "App fallback metadata detected"
+            # Reject app fallbacks
-            # must contain canonical ExifTool keys
+            assert "no-metadata" not in html and "exif_err:" not in html, "App fallback metadata detected"
-            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")
+            # Required keys appear anywhere in the HTML
+            assert "File Name:" in html, "Missing 'File Name:' in post HTML"
+            assert "Date Created:" in html, "Missing 'Date Created:' in post HTML"
+
+            self.logger.info("    ✓ Required metadata keys found in page HTML")
         except Exception as e:
             self.logger.error(f"    ✗ Metadata verification failed: {e}")
             return False

services/blogpost/chall/app.py

@@ -207,19 +207,29 @@ def create_post():
 
 @app.route('/uploads/<path:filename>')
 def uploaded_file(filename):
+    if "user_id" not in session:
+        return redirect(url_for("login"))
+    if ".." in filename or filename.startswith("/"):
+        abort(404)
+    db = get_db()
+    cur = db.execute("SELECT id FROM posts WHERE image_filename = ? AND author_id = ?", (filename, session["user_id"]))
+    row = cur.fetchone()
+    if not row:
+        abort(404)
     return send_from_directory(app.config['UPLOAD_FOLDER'], filename)
 
 @app.route("/post/<int:pid>")
 def view_post(pid):
+    if "user_id" not in session:
+        return redirect(url_for("login"))
     db = get_db()
     cur = db.execute(
-        "SELECT p.*, u.username as author FROM posts p LEFT JOIN users u ON p.author_id = u.id WHERE p.id = ?",
+        "SELECT p.*, u.username as author FROM posts p LEFT JOIN users u ON p.author_id = u.id WHERE p.id = ? AND p.author_id = ?",
-        (pid,)
+        (pid, session["user_id"])
     )
     post = cur.fetchone()
     if not post:
         abort(404)
-
     return render_template("view_post.html", post=post)
 
 @app.route("/profile")

services/cdn/chall/app.py

@@ -6,6 +6,8 @@ import secrets
 import datetime
 import subprocess
 from pathlib import Path
+from functools import wraps
+
 from flask import *
 from werkzeug.security import generate_password_hash, check_password_hash
 from werkzeug.utils import secure_filename
@@ -13,7 +15,6 @@ from werkzeug.utils import secure_filename
 APP_DIR = os.path.dirname(os.path.abspath(__file__))
 DB_PATH = os.path.join(APP_DIR, "data.db")
 UPLOAD_DIR = os.path.join(APP_DIR, "uploads")
-FLAG_PATH = os.path.join("/flag.txt")
 
 ALLOWED_EXT = {"png", "jpg", "jpeg", "bmp"}
 MAX_CONTENT_LENGTH = 8 * 1024 * 1024
@@ -39,7 +40,8 @@ def close_db(_exc):
 def init_db():
     Path(UPLOAD_DIR).mkdir(parents=True, exist_ok=True)
     db = get_db()
-    db.executescript("""
+    db.executescript(
+        """
     CREATE TABLE IF NOT EXISTS users (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         username TEXT UNIQUE NOT NULL,
@@ -57,24 +59,12 @@ def init_db():
         created_at TEXT NOT NULL,
         FOREIGN KEY(user_id) REFERENCES users(id)
     );
-    """)
+
+    CREATE INDEX IF NOT EXISTS idx_posts_user_file ON posts(user_id, filename);
+    """
+    )
     db.commit()
 
-def _resolve_flag_file_path():
-    p = FLAG_PATH
-    if os.path.isdir(p):
-        p = os.path.join(p, "flag.txt")
-    os.makedirs(os.path.dirname(p), exist_ok=True)
-    return p
-
-def generate_flag_at_boot():
-    path = _resolve_flag_file_path()
-    if not os.path.exists(path) or os.environ.get("RESEED_FLAG") == "1":
-        token = secrets.token_bytes(32)
-        sha = hashlib.sha256(token).hexdigest()
-        with open(path, "w", encoding="utf-8") as fh:
-            fh.write(f"GEMASTIK{{{sha}}}\n")
-
 def current_user():
     if "uid" not in session:
         return None
@@ -82,6 +72,16 @@ def current_user():
     cur = db.execute("SELECT id, username, role FROM users WHERE id = ?", (session["uid"],))
     return cur.fetchone()
 
+def login_required(view):
+    @wraps(view)
+    def wrapped(*args, **kwargs):
+        if not current_user():
+            dest = request.path
+            flash("Please log in to continue.")
+            return redirect(url_for("login", next=dest))
+        return view(*args, **kwargs)
+    return wrapped
+
 def _is_within(child_path: str, parent_dir: str) -> bool:
     child_real = os.path.realpath(child_path)
     parent_real = os.path.realpath(parent_dir)
@@ -156,8 +156,13 @@ def login():
         if not row or not check_password_hash(row["password_hash"], password):
             flash("Invalid credentials")
             return render_template("login.html")
+
         session["uid"] = row["id"]
         flash(f"Welcome, {row['username']}!")
+
+        next_url = request.args.get("next") or request.form.get("next")
+        if next_url and next_url.startswith("/"):
+            return redirect(next_url)
         return redirect(url_for("gallery"))
     return render_template("login.html")
 
@@ -168,10 +173,9 @@ def logout():
     return redirect(url_for("login"))
 
 @app.route("/upload", methods=["GET", "POST"])
+@login_required
 def upload():
     user = current_user()
-    if not user:
-        return redirect(url_for("login"))
     if request.method == "POST":
         title = request.form.get("title", "").strip() or "(untitled)"
         f = request.files.get("image")
@@ -205,6 +209,7 @@ def upload():
 
 @app.route("/")
 @app.route("/gallery")
+@login_required
 def gallery():
     user = current_user()
     db = get_db()
@@ -212,18 +217,22 @@ def gallery():
         """
         SELECT p.id, p.title, p.filename, p.created_at, u.username
         FROM posts p JOIN users u ON p.user_id = u.id
+        WHERE p.user_id = ?
         ORDER BY p.id DESC
-        """
+        """,
+        (user["id"],),
     )
     posts = cur.fetchall()
     return render_template("gallery.html", posts=posts, user=user)
 
 @app.route("/post/<int:pid>")
+@login_required
 def view_post(pid):
+    user = current_user()
     db = get_db()
     cur = db.execute(
-        "SELECT p.*, u.username FROM posts p JOIN users u ON p.user_id = u.id WHERE p.id = ?",
+        "SELECT p.*, u.username FROM posts p JOIN users u ON p.user_id = u.id WHERE p.id = ? AND p.user_id = ?",
-        (pid,),
+        (pid, user["id"]),
     )
     post = cur.fetchone()
     if not post:
@@ -250,8 +259,20 @@ def view_post(pid):
     return render_template_string(page_src, post=post)
 
 @app.route("/i/<path:filename>")
+@login_required
 def cdn_serve(filename):
-    return send_from_directory(UPLOAD_DIR, filename, as_attachment=False)
+    user = current_user()
+    db = get_db()
+    cur = db.execute(
+        "SELECT 1 FROM posts WHERE user_id = ? AND filename = ? LIMIT 1",
+        (user["id"], filename),
+    )
+    if not cur.fetchone():
+        abort(404)  
+
+    resp = send_from_directory(UPLOAD_DIR, filename, as_attachment=False)
+    resp.headers["Cache-Control"] = "private, max-age=0, no-store"
+    return resp
 
 @app.errorhandler(413)
 def too_large(_):