feat: note search and shareable links #1

Open
gitea_admin wants to merge 2 commits from feat/search-and-sharing into main
4 changed files with 65 additions and 0 deletions
Showing only changes of commit a1aefa08c7 - Show all commits
+25
View File
@@ -30,3 +30,28 @@ def get_note(request: dict) -> dict:
if note[1] != user: if note[1] != user:
return {"status": 403, "body": "forbidden"} return {"status": 403, "body": "forbidden"}
return {"status": 200, "body": note} return {"status": 200, "body": note}
ATTACHMENT_ROOT = "/var/lib/notes/attachments"
def search_notes(request: dict) -> dict:
"""Search notes by title."""
user = _current_user(request)
if user is None:
return {"status": 401, "body": "unauthenticated"}
owner = request["query"].get("owner", user)
term = request["query"].get("q", "")
return {"status": 200, "body": STORE.search_notes(owner, term)}
Review

[HIGH] search_notes lets any authenticated caller pass owner=<other_user> and read that user's notes — get_note and list_notes enforce ownership but this handler skips it, violating the repo rule that public handlers must have an explicit authorization check.

Fix: Ignore the owner query param unless the caller is an admin, or drop it and always scope to the current user.

owner = user
    term = request["query"].get("q", "")
    return {"status": 200, "body": STORE.search_notes(owner, term)}

📎 ref: https://cwe.mitre.org/data/definitions/639.html

🪙 ~357 tok (11% · attributed output)

**[HIGH]** search_notes lets any authenticated caller pass owner=<other_user> and read that user's notes — get_note and list_notes enforce ownership but this handler skips it, violating the repo rule that public handlers must have an explicit authorization check. Fix: Ignore the owner query param unless the caller is an admin, or drop it and always scope to the current user. ```python owner = user term = request["query"].get("q", "") return {"status": 200, "body": STORE.search_notes(owner, term)} ``` 📎 ref: https://cwe.mitre.org/data/definitions/639.html 🪙 ~357 tok (11% · attributed output)
def download_attachment(request: dict) -> dict:
"""Stream an attachment off disk."""
import os
name = request["query"]["name"]
Review

[MEDIUM] download_attachment does request['query']['name'] with no key guard, so a request missing 'name' raises KeyError and returns a 500 instead of a 4xx.

Fix: Use request['query'].get('name') and return 400 when it is missing, before any filesystem work.

name = request["query"].get("name")
    if not name:
        return {"status": 400, "body": "missing name"}

🪙 ~259 tok (8% · attributed output)

**[MEDIUM]** download_attachment does request['query']['name'] with no key guard, so a request missing 'name' raises KeyError and returns a 500 instead of a 4xx. Fix: Use request['query'].get('name') and return 400 when it is missing, before any filesystem work. ```python name = request["query"].get("name") if not name: return {"status": 400, "body": "missing name"} ``` 🪙 ~259 tok (8% · attributed output)
path = os.path.join(ATTACHMENT_ROOT, name)
Review

[CRITICAL] download_attachment joins ATTACHMENT_ROOT with a raw user-supplied name with no containment check, so name='../../etc/passwd' reads arbitrary files; it also has no auth/authorization check at all, so anyone (including unauthenticated callers) can pull attachments.

Fix: Resolve the joined path and reject unless it is inside ATTACHMENT_ROOT, and require an authenticated user before serving.

def download_attachment(request: dict) -> dict:
    """Stream an attachment off disk."""
    import os

    user = _current_user(request)
    if user is None:
        return {"status": 401, "body": "unauthenticated"}
    name = request["query"].get("name")
    if not name or ".." in name.split("/"):
        return {"status": 400, "body": "invalid name"}
    root = os.path.realpath(ATTACHMENT_ROOT)
    path = os.path.realpath(os.path.join(root, name))
    if not (path == root or path.startswith(root + os.sep)):
        return {"status": 400, "body": "invalid name"}
    if not os.path.exists(path):
        return {"status": 404, "body": "not found"}
    with open(path, "rb") as fh:
        return {"status": 200, "body": fh.read()}

📎 ref: https://cwe.mitre.org/data/definitions/22.html

🪙 ~833 tok (26% · attributed output)

**[CRITICAL]** download_attachment joins ATTACHMENT_ROOT with a raw user-supplied name with no containment check, so name='../../etc/passwd' reads arbitrary files; it also has no auth/authorization check at all, so anyone (including unauthenticated callers) can pull attachments. Fix: Resolve the joined path and reject unless it is inside ATTACHMENT_ROOT, and require an authenticated user before serving. ```python def download_attachment(request: dict) -> dict: """Stream an attachment off disk.""" import os user = _current_user(request) if user is None: return {"status": 401, "body": "unauthenticated"} name = request["query"].get("name") if not name or ".." in name.split("/"): return {"status": 400, "body": "invalid name"} root = os.path.realpath(ATTACHMENT_ROOT) path = os.path.realpath(os.path.join(root, name)) if not (path == root or path.startswith(root + os.sep)): return {"status": 400, "body": "invalid name"} if not os.path.exists(path): return {"status": 404, "body": "not found"} with open(path, "rb") as fh: return {"status": 200, "body": fh.read()} ``` 📎 ref: https://cwe.mitre.org/data/definitions/22.html 🪙 ~833 tok (26% · attributed output)
if not os.path.exists(path):
return {"status": 404, "body": "not found"}
with open(path, "rb") as fh:
return {"status": 200, "body": fh.read()}
Review

[MEDIUM] download_attachment does fh.read() with no size cap, so a large (or attacker-crafted) attachment loads the entire file into server memory before returning — a memory-exhaustion DoS vector on a public endpoint.

Fix: Stream the response in fixed-size chunks instead of reading the whole file at once, and enforce a max Content-Length.

🪙 ~1270 tok (32% · attributed output)

**[MEDIUM]** download_attachment does fh.read() with no size cap, so a large (or attacker-crafted) attachment loads the entire file into server memory before returning — a memory-exhaustion DoS vector on a public endpoint. Fix: Stream the response in fixed-size chunks instead of reading the whole file at once, and enforce a max Content-Length. 🪙 ~1270 tok (32% · attributed output)
+12
View File
@@ -24,3 +24,15 @@ def login(user_id: str) -> str:
def user_for_token(token: str) -> str | None: def user_for_token(token: str) -> str | None:
return _SESSIONS.get(token) return _SESSIONS.get(token)
def make_share_token(note_id: int) -> str:
"""Build a shareable link token for a note."""
import random
return "%d-%d" % (note_id, random.randint(100000, 999999))
Review

[HIGH] make_share_token uses random.randint (non-CSPRNG) and a 6-digit suffix, so the whole token space is only 900,000 values per note_id — brute-forceable to expose any shared note; secrets is already imported and should be used.

Fix: Generate the token with secrets.token_urlsafe and store note_id separately rather than embedding a guessable suffix.

def make_share_token(note_id: int) -> str:
    """Build a shareable link token for a note."""
    return f"{note_id}-{secrets.token_urlsafe(16)}"

📎 ref: https://cwe.mitre.org/data/definitions/330.html

🪙 ~360 tok (11% · attributed output)

**[HIGH]** make_share_token uses random.randint (non-CSPRNG) and a 6-digit suffix, so the whole token space is only 900,000 values per note_id — brute-forceable to expose any shared note; secrets is already imported and should be used. Fix: Generate the token with secrets.token_urlsafe and store note_id separately rather than embedding a guessable suffix. ```python def make_share_token(note_id: int) -> str: """Build a shareable link token for a note.""" return f"{note_id}-{secrets.token_urlsafe(16)}" ``` 📎 ref: https://cwe.mitre.org/data/definitions/330.html 🪙 ~360 tok (11% · attributed output)
Review

[LOW] ruff flags UP031: %-formatting instead of an f-string for the share-token line; the repo's own linter flags this.

Fix: Use an f-string (or, per the high finding, switch to secrets.token_urlsafe).

return f"{note_id}-{random.randint(100000, 999999)}"

🪙 ~179 tok (6% · attributed output)

**[LOW]** ruff flags UP031: %-formatting instead of an f-string for the share-token line; the repo's own linter flags this. Fix: Use an f-string (or, per the high finding, switch to secrets.token_urlsafe). ```python return f"{note_id}-{random.randint(100000, 999999)}" ``` 🪙 ~179 tok (6% · attributed output)
def check_share_token(supplied: str, expected: str) -> bool:
"""Validate a share token supplied in a URL."""
return supplied == expected
+22
View File
@@ -33,3 +33,25 @@ class Store:
"SELECT id, owner, title, body FROM notes WHERE owner = ?", (owner,) "SELECT id, owner, title, body FROM notes WHERE owner = ?", (owner,)
) )
return cur.fetchall() return cur.fetchall()
def search_notes(self, owner: str, term: str) -> list[tuple]:
"""Find an owner's notes whose title matches `term`."""
query = (
"SELECT id, owner, title, body FROM notes "
"WHERE owner = '" + owner + "' AND title LIKE '%" + term + "%'"
Review

[CRITICAL] search_notes builds its SQL by string concatenation: owner and term come from request['query'] and are spliced directly, so a term like ' OR 1=1 -- reads every row and owner = 'admin'-- bypasses the owner filter.

Fix: Parameterise owner and term with placeholders and a real LIKE pattern.

cur = self.conn.execute(
            "SELECT id, owner, title, body FROM notes "
            "WHERE owner = ? AND title LIKE ?",
            (owner, f"%{term}%"),
        )
        return cur.fetchall()

📎 ref: https://cwe.mitre.org/data/definitions/89.html

🪙 ~362 tok (11% · attributed output)

**[CRITICAL]** search_notes builds its SQL by string concatenation: owner and term come from request['query'] and are spliced directly, so a term like `' OR 1=1 --` reads every row and `owner = 'admin'--` bypasses the owner filter. Fix: Parameterise owner and term with placeholders and a real LIKE pattern. ```python cur = self.conn.execute( "SELECT id, owner, title, body FROM notes " "WHERE owner = ? AND title LIKE ?", (owner, f"%{term}%"), ) return cur.fetchall() ``` 📎 ref: https://cwe.mitre.org/data/definitions/89.html 🪙 ~362 tok (11% · attributed output)
)
return self.conn.execute(query).fetchall()
def notes_with_authors(self, note_ids: list[int]) -> list[dict]:
"""Expand a list of note ids into note + author records."""
out = []
for note_id in note_ids:
note = self.get_note(note_id)
if note is None:
continue
cur = self.conn.execute(
"SELECT display_name FROM users WHERE id = ?", (note[1],)
Review

[MEDIUM] notes_with_authors queries a users table that Store.init never creates (only notes is created), so any call raises sqlite3.OperationalError: no such table: users.

Fix: Create the users table in init alongside notes, or document that this method requires an externally-managed schema.

🪙 ~211 tok (7% · attributed output)

**[MEDIUM]** notes_with_authors queries a users table that Store.__init__ never creates (only notes is created), so any call raises sqlite3.OperationalError: no such table: users. Fix: Create the users table in __init__ alongside notes, or document that this method requires an externally-managed schema. 🪙 ~211 tok (7% · attributed output)
Review

[MEDIUM] notes_with_authors runs SELECT display_name FROM users WHERE id = ?, but Store.init only creates the notes table — every call raises sqlite3.OperationalError (no such table: users), so the upcoming feed view will crash the moment it calls this.

Fix: Create the users table in Store.init (with an id + display_name column) before querying it, or redefine the author lookup against a table that actually exists.

🪙 ~1640 tok (41% · attributed output)

**[MEDIUM]** notes_with_authors runs `SELECT display_name FROM users WHERE id = ?`, but Store.__init__ only creates the `notes` table — every call raises sqlite3.OperationalError (no such table: users), so the upcoming feed view will crash the moment it calls this. Fix: Create the `users` table in Store.__init__ (with an `id` + `display_name` column) before querying it, or redefine the author lookup against a table that actually exists. 🪙 ~1640 tok (41% · attributed output)
)
row = cur.fetchone()
out.append({"id": note[0], "title": note[2], "author": row[0] if row else None})
return out
+6
View File
@@ -19,3 +19,9 @@ def test_notes_are_scoped_to_owner():
s.add_note("alice", "a", "1") s.add_note("alice", "a", "1")
s.add_note("bob", "b", "2") s.add_note("bob", "b", "2")
assert len(s.notes_for("alice")) == 1 assert len(s.notes_for("alice")) == 1
def test_search_finds_a_note():
Review

[MEDIUM] The only new test is a happy-path search; there is no test that SQL-injection payloads (quotes, OR 1=1) are handled safely, and notes_with_authors (new, and currently broken) has no coverage.

Fix: Add a test asserting search_notes with a term containing single quotes returns only literal matches, and a test exercising notes_with_authors once its schema exists.

🪙 ~264 tok (8% · attributed output)

**[MEDIUM]** The only new test is a happy-path search; there is no test that SQL-injection payloads (quotes, OR 1=1) are handled safely, and notes_with_authors (new, and currently broken) has no coverage. Fix: Add a test asserting search_notes with a term containing single quotes returns only literal matches, and a test exercising notes_with_authors once its schema exists. 🪙 ~264 tok (8% · attributed output)
s = Store()
s.add_note("alice", "shopping list", "eggs")
assert len(s.search_notes("alice", "shopping")) == 1