Initial commit

2025-10-10 22:18:06 +07:00
commit ea02892f14
1246 changed files with 289342 additions and 0 deletions
@@ -0,0 +1,2 @@
+__pycache__/
+*.db
@@ -0,0 +1 @@
+9-<-J:@JrI;g;+oY!:
@@ -0,0 +1,24 @@
+FROM public.ecr.aws/docker/library/python:latest
+
+ARG PASSWORD
+
+RUN echo root:${PASSWORD} | chpasswd
+RUN apt-get update && apt-get install -y openssh-server sqlite3 curl
+RUN echo "PasswordAuthentication yes" >> /etc/ssh/sshd_config
+RUN echo "PermitRootLogin yes" >> /etc/ssh/sshd_config
+RUN service ssh start
+
+WORKDIR /ctf/pasta/
+
+COPY src/ src/
+COPY requirements.txt .
+COPY start.sh .
+
+RUN pip install -r requirements.txt
+RUN touch flag.txt
+RUN python3 src/initdb.py deomkicer_number_1_fans ${PASSWORD}
+
+RUN chmod +x start.sh
+CMD ./start.sh
+
+# docker build --build-arg "PASSWORD=123" -t pasta .
@@ -0,0 +1 @@
+PLACEHOLDER
@@ -0,0 +1,5 @@
+fastapi>=0.68.0,<0.69.0
+pydantic>=1.8.0,<2.0.0
+uvicorn>=0.15.0,<0.16.0
+ecdsa
+sqlalchemy
@@ -0,0 +1,111 @@
+import os
+import json
+from hashlib import sha256
+from fastapi import FastAPI, Depends, Response, Request
+from sqlalchemy.orm import Session
+from pydantic import BaseModel
+from database import get_db
+from pasta import PastaSigner, PastaVerifier
+import models
+
+
+class Credential(BaseModel):
+    username: str
+    password: str
+    role: str | None = 'user'
+
+
+secret = b'\x00' + os.urandom(65)
+
+app = FastAPI()
+
+
+def auth(token):
+    if token:
+        splitted = token.split(" ")
+        if len(splitted) == 2 and splitted[0] == "Bearer":
+            return splitted[1]
+
+    return False
+
+
+@app.get('/')
+def index(request: Request):
+    token = auth(request.headers.get('Authorization'))
+    if not token:
+        return {"message": "Welcome to PASTA: Platform-Agnostic Security Token for Authentication"}
+
+    verifier = PastaVerifier(secret)
+
+    try:
+        data = verifier.verify(token)
+        if not data:
+            return Response(json.dumps({"error": "Invalid token"}), 401)
+
+        return data
+    except Exception:
+        return Response(json.dumps({"error": "Invalid token"}), 401)
+
+
+@app.get('/flag')
+def flag(request: Request):
+    token = auth(request.headers.get('Authorization'))
+    if not token:
+        return Response(json.dumps({"error": "Invalid token"}), 401)
+
+    verifier = PastaVerifier(secret)
+
+    try:
+        data = verifier.verify(token)
+        if not data:
+            return Response(json.dumps({"error": "Invalid token"}), 401)
+
+        if data['role'] != 'admin':
+            return Response(json.dumps({"error": "Only admin can see the flag"}), 403)
+
+        flag = open('../flag.txt', 'r').read()
+        return {"flag": flag}
+    except Exception:
+        return Response(json.dumps({"error": "Invalid token"}), 401)
+
+
+@app.post("/auth")
+def login(credential: Credential, version: int | None = 1, db: Session = Depends(get_db)):
+    users = db.query(models.User).filter(
+        models.User.username == credential.username,
+        models.User.password == sha256(credential.password.encode()).hexdigest()
+    ).all()
+
+    if len(users) != 1:
+        return Response(json.dumps({"error": "Username/password is incorrect"}), 401)
+
+    data = json.dumps({
+        "username": users[0].username,
+        "role": users[0].role
+    })
+
+    signer = PastaSigner(secret, version)
+    token = signer.sign(data)
+
+    return {"token": token}
+
+
+@app.post("/register")
+def register(credential: Credential, db: Session = Depends(get_db)):
+
+    users = db.query(models.User).filter(
+        models.User.username == credential.username
+    ).all()
+
+    if len(users) == 0:
+        credential.password = sha256(credential.password.encode()).hexdigest()
+        credential.role = "user"
+        new_user = models.User(**credential.dict())
+
+        db.add(new_user)
+        db.commit()
+        db.refresh(new_user)
+
+        return Response(json.dumps({"success": "User registered succesfully"}), 200)
+    else:
+        return Response(json.dumps({"error": "Username already exist"}), 409)
@@ -0,0 +1,20 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+SQLITE_DATABASE_URL = "sqlite:///./pasta.db"
+
+engine = create_engine(
+    SQLITE_DATABASE_URL, echo=True, connect_args={"check_same_thread": False}
+)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+import hashlib
+import os
+import sqlite3
+import sys
+
+db = "src/pasta.db"
+
+
+class User:
+    def __init__(self, username, password, role):
+        self.username: str = username
+        self.password: str = password
+        self.role: str = role
+
+
+def main():
+    try:
+        os.remove(db)
+    except BaseException:
+        pass
+
+    con = sqlite3.connect(db)
+    con.execute("create table users(username varchar(200) primary key,password varchar(64),role varchar(30) default 'user')")
+
+    def insert_person(person: User):
+        cur.execute(
+            "INSERT INTO users (username, password, role) VALUES (?, ?, ?)",
+            (
+                person.username,
+                hashlib.sha256(
+                    person.password.encode()
+                ).hexdigest(),
+                person.role,
+            ),
+        )
+        con.commit()
+
+    cur = con.cursor()
+    admin = User(sys.argv[1], sys.argv[2], "admin")
+    insert_person(admin)
+    con.close()
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 3:
+        print(f"Usage: {sys.argv[0]} <username> <password>")
+        exit()
+    main()
@@ -0,0 +1,9 @@
+from database import Base
+from sqlalchemy import Column, String, Boolean
+
+
+class User(Base):
+    __tablename__ = 'users'
+    username = Column(String, primary_key=True, nullable=False)
+    password = Column(String, nullable=False)
+    role = Column(String, nullable=True)
@@ -0,0 +1,88 @@
+import hmac
+import json
+from base64 import urlsafe_b64encode, urlsafe_b64decode
+from hashlib import sha224, sha256, sha384, sha512
+from ecdsa import SigningKey, NIST256p, NIST224p, NIST384p, NIST521p
+
+allowed_curve = [
+    NIST224p,
+    NIST256p,
+    NIST384p,
+    NIST521p
+]
+
+hashfunc = [
+    sha224,
+    sha256,
+    sha384,
+    sha512
+]
+
+key_size = [28, 32, 48, 66]
+signature_size = [56, 64, 96, 132]
+
+
+class PastaSigner:
+    def __init__(self, secret: bytes, version: int):
+        self.purpose = 'public'
+        if version > 4 or version < 1:
+            version = 1
+        self.version = version
+        self.hashfunc = hashfunc[self.version - 1]
+        self.key_size = key_size[self.version - 1]
+        self.priv = SigningKey.from_string(secret[:self.key_size], curve=allowed_curve[self.version - 1])
+
+    def serialize(self, data: bytes, sig):
+        token = 'v' + str(self.version) + '.'
+        token += self.purpose + '.'
+        token += urlsafe_b64encode(data + sig).decode().replace('=', '')
+        return token
+
+    def sign(self, data: str):
+        data = data.encode()
+        pub = self.priv.get_verifying_key().to_string()
+        h = self.hashfunc(data + pub).digest()
+        nonce = hmac.new(self.priv.to_string(), data, self.hashfunc).hexdigest()
+        sig = self.priv.sign_digest(h, k=int(nonce, 16))
+
+        return self.serialize(data + pub, sig)
+
+
+class PastaVerifier:
+
+    def __init__(self, secret):
+        self.purpose = 'public'
+        self.secret = secret
+
+    def deserialize(self, data: bytes):
+        try:
+            version, purpose, payload = data.split(b'.')
+            version = int(version.replace(b'v', b''))
+            if version > 4 or version < 1:
+                return False
+            self.version = version
+            self.hashfunc = hashfunc[self.version - 1]
+            self.key_size = key_size[self.version - 1]
+            self.priv = SigningKey.from_string(self.secret[:self.key_size], curve=allowed_curve[self.version - 1])
+
+            raw_data = urlsafe_b64decode(payload + (b'==' * 2))
+            size = signature_size[self.version - 1]
+            signature = raw_data[-size:]
+            public_key = raw_data[-size * 2:-size]
+            message = raw_data[:-size]
+
+            return message, raw_data[:-size * 2], signature
+        except Exception as e:
+            return False
+
+    def verify(self, token: str):
+        deserialized = self.deserialize(token.encode())
+        if deserialized:
+            message, plain_data, signature = deserialized
+            h = self.hashfunc(message).digest()
+            verifier = self.priv.get_verifying_key()
+            verifier.verify_digest(signature, h)
+
+            return json.loads(plain_data)
+
+        return False
@@ -0,0 +1,5 @@
+# run sshd
+/usr/sbin/sshd -D &
+# run the command
+cd src
+uvicorn app:app --host 0.0.0.0 --port 8000