initial: toy notes service used to demo pragent
A deliberately small Python service — auth helpers, a sqlite-backed store, and two request handlers — plus a .pr-review.json that steers the reviewer toward this repo's house rules. Pull requests against it carry planted defects so the reviewer has something real to find.
This commit is contained in:
+32
@@ -0,0 +1,32 @@
|
||||
"""Request handlers. `request` is a dict of {path, query, headers}."""
|
||||
|
||||
from app import auth
|
||||
from app.store import Store
|
||||
|
||||
STORE = Store()
|
||||
|
||||
|
||||
def _current_user(request: dict) -> str | None:
|
||||
token = request.get("headers", {}).get("Authorization", "").removeprefix("Bearer ")
|
||||
if not token:
|
||||
return None
|
||||
return auth.user_for_token(token)
|
||||
|
||||
|
||||
def list_notes(request: dict) -> dict:
|
||||
user = _current_user(request)
|
||||
if user is None:
|
||||
return {"status": 401, "body": "unauthenticated"}
|
||||
return {"status": 200, "body": STORE.notes_for(user)}
|
||||
|
||||
|
||||
def get_note(request: dict) -> dict:
|
||||
user = _current_user(request)
|
||||
if user is None:
|
||||
return {"status": 401, "body": "unauthenticated"}
|
||||
note = STORE.get_note(int(request["query"]["id"]))
|
||||
if note is None:
|
||||
return {"status": 404, "body": "not found"}
|
||||
if note[1] != user:
|
||||
return {"status": 403, "body": "forbidden"}
|
||||
return {"status": 200, "body": note}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
"""Token and session helpers."""
|
||||
|
||||
import hmac
|
||||
import secrets
|
||||
|
||||
_SESSIONS: dict[str, str] = {}
|
||||
|
||||
|
||||
def new_token() -> str:
|
||||
"""Generate an opaque session token."""
|
||||
return secrets.token_urlsafe(32)
|
||||
|
||||
|
||||
def verify_token(supplied: str, expected: str) -> bool:
|
||||
"""Constant-time token comparison."""
|
||||
return hmac.compare_digest(supplied, expected)
|
||||
|
||||
|
||||
def login(user_id: str) -> str:
|
||||
token = new_token()
|
||||
_SESSIONS[token] = user_id
|
||||
return token
|
||||
|
||||
|
||||
def user_for_token(token: str) -> str | None:
|
||||
return _SESSIONS.get(token)
|
||||
@@ -0,0 +1,35 @@
|
||||
"""A toy data store. Pretend it is a real database."""
|
||||
|
||||
import sqlite3
|
||||
|
||||
|
||||
class Store:
|
||||
def __init__(self, path: str = ":memory:"):
|
||||
self.conn = sqlite3.connect(path, check_same_thread=False)
|
||||
self.conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS notes ("
|
||||
" id INTEGER PRIMARY KEY,"
|
||||
" owner TEXT NOT NULL,"
|
||||
" title TEXT NOT NULL,"
|
||||
" body TEXT NOT NULL)"
|
||||
)
|
||||
|
||||
def add_note(self, owner: str, title: str, body: str) -> int:
|
||||
cur = self.conn.execute(
|
||||
"INSERT INTO notes (owner, title, body) VALUES (?, ?, ?)",
|
||||
(owner, title, body),
|
||||
)
|
||||
self.conn.commit()
|
||||
return cur.lastrowid
|
||||
|
||||
def get_note(self, note_id: int) -> tuple | None:
|
||||
cur = self.conn.execute(
|
||||
"SELECT id, owner, title, body FROM notes WHERE id = ?", (note_id,)
|
||||
)
|
||||
return cur.fetchone()
|
||||
|
||||
def notes_for(self, owner: str) -> list[tuple]:
|
||||
cur = self.conn.execute(
|
||||
"SELECT id, owner, title, body FROM notes WHERE owner = ?", (owner,)
|
||||
)
|
||||
return cur.fetchall()
|
||||
Reference in New Issue
Block a user