From 0472ee1dd3eb075690783864d541408d6c09bd12 Mon Sep 17 00:00:00 2001 From: Ignace Date: Thu, 14 May 2026 14:30:23 +0200 Subject: [PATCH] group bugs and authorization improvements by codex --- api.py | 47 ++++++++++++++++++++--------------------- auth.py | 25 +++++++++++++++------- items.py | 32 +++++++++++++++++++++------- lists.py | 23 ++++++++++++++------ models.py | 16 +++++++++++--- templates/register.html | 5 +++++ 6 files changed, 98 insertions(+), 50 deletions(-) diff --git a/api.py b/api.py index 57f1a80..36b397a 100644 --- a/api.py +++ b/api.py @@ -1,5 +1,5 @@ from flask import Blueprint, render_template, redirect, url_for, request -from flask_login import login_required, current_user +from flask_login import login_required, current_user, login_user from models import db from models import User, Item import json @@ -14,7 +14,6 @@ def validate_key(): result['data'] = '' if "key" in request.form: key = request.form["key"] - print(key) if key == 'NogNietNodigHier': result['status'] = 1 else: @@ -29,12 +28,14 @@ def login(): if "user" in request.form and "password" in request.form: user = request.form["user"] password = request.form["password"] - print(user) user = User.query.filter_by(name=user).first() if user and user.check_password(password): if not user.is_approved: result['message'] = 'User is pending approval.' + elif user.group_id == 0: + result['message'] = 'User is not connected to a group.' else: + login_user(user) result['status'] = 1 result['data'] = user.as_dict() else: @@ -47,53 +48,51 @@ def register(): result['status']=0 result['data'] = '' if "user" in request.form and "password" in request.form: - username = request.form["user"] + username = request.form["user"].strip() password = request.form["password"] - print(username) user = User.query.filter_by(name=username).first() if user: result['message'] = 'Username is not unique. Pick something else please.' else: + user = User(name=username) + group_id = user.set_group_id(password) + if group_id == 0: + result['message'] = 'Unknown registration key.' + return json.dumps(result), 200 + result['status'] = 2 result['message'] = 'Registration Ok. Awaiting approval.' - user = User(name=username) user.set_password(password) + user.group_id = group_id db.session.add(user) db.session.commit() return json.dumps(result), 200 @api_bp.route("/load_lists", methods=["POST"]) +@login_required def load_lists(): - print("Trying to load lists") result={} result['status']=1 # from this call the result is always 1, empty or not - result['data'] = '' - if "userid" in request.form: - userid = int(request.form["userid"]) - - print(userid) - user = User.query.filter_by(id=userid).first() - if user: - result['data'] = user.all_my_lists_as_dict() + result['data'] = current_user.all_my_lists_as_dict() return json.dumps(result), 200 @api_bp.route("/load_items", methods=["POST"]) +@login_required def load_items(): - print("Trying to load items") result={} result['status']=1 # from this call the result is always 1, empty or not result['data'] = '' if "listid" in request.form: listid = int(request.form["listid"]) + if current_user.can_access_list(listid): + items=Item.query.filter_by(listofitems_id=listid, is_suggestion=False).order_by(text("is_checked, category, label")).all() + if items: + result['data'] = [ i.as_dict() for i in items] + else: + result['status'] = 0 + result['message'] = 'Forbidden' + return json.dumps(result), 403 - print(listid) - items=Item.query.filter_by(listofitems_id=listid, is_suggestion=False).order_by(text("is_checked, category, label")).all() - print(items) - if items: - result['data'] = [ i.as_dict() for i in items] - - print(result) return json.dumps(result), 200 - diff --git a/auth.py b/auth.py index 227b26a..620ce75 100644 --- a/auth.py +++ b/auth.py @@ -93,13 +93,22 @@ def login(): @auth_bp.route("/register", methods=["GET", "POST"]) def register(): if request.method == "POST": - user = User(name=request.form["name"]) - user.set_password(request.form["password"]) - group_id = user.get_group_id(request.form["password"]) - if group_id > 0: - user.group_id = group_id - db.session.add(user) - db.session.commit() + name = request.form["name"].strip() + password = request.form["password"] + + existing_user = User.query.filter_by(name=name).first() + if existing_user: + return render_template("register.html", error="That name is already registered.") + + user = User(name=name) + group_id = user.set_group_id(password) + if group_id == 0: + return render_template("register.html", error="Unknown registration key.") + + user.set_password(password) + user.group_id = group_id + db.session.add(user) + db.session.commit() return render_template("pending.html") return render_template("register.html") @@ -130,4 +139,4 @@ def change_pwd(): Log.info(f"Authorization failure for user '{request.form["name"]}' ") return render_template(url_for("lists.home")) - return render_template("change_pwd.html") \ No newline at end of file + return render_template("change_pwd.html") diff --git a/items.py b/items.py index 96d8c8b..7702fd3 100644 --- a/items.py +++ b/items.py @@ -1,7 +1,7 @@ -from flask import Blueprint, render_template, redirect, url_for, request, make_response +from flask import Blueprint, render_template, redirect, url_for, request, make_response, abort from flask_login import login_required, current_user from models import db -from models import Item +from models import Item, ListOfItems from sqlalchemy import desc, text import os from log import Log @@ -11,10 +11,23 @@ items_bp = Blueprint("items", __name__) ICONPATH = "static/categories/" #without starting / and with ending / +def require_list_access(listid): + listofitem = db.get_or_404(ListOfItems, int(listid)) + if not current_user.can_access_list(listofitem.id): + abort(403) + return listofitem + +def require_item_access(itemid): + item = db.get_or_404(Item, int(itemid)) + if not current_user.can_access_item(item.id): + abort(403) + return item + # home screen for items @items_bp.route("/items/") @login_required def items(listid): + require_list_access(listid) resp = make_response(render_template("items_show.html", user=current_user, items=Item.query.filter_by(listofitems_id=listid, is_suggestion=False).order_by(text("is_checked, category, label")).all(), button_top_url1 = url_for("items.items_append", listid=listid), @@ -28,7 +41,7 @@ def items(listid): @items_bp.route("/item_update//") @login_required def item_update(itemid, itemchecked): - i = Item.query.get(itemid) + i = require_item_access(itemid) i.is_checked = True if itemchecked == 'true' else False db.session.commit() return '', 204 @@ -115,6 +128,7 @@ def interpret_and_add_item(listid, newlabel): @items_bp.route("/items_append/", methods=["GET", "POST"]) @login_required def items_append(listid): + require_list_access(listid) if request.method == "POST": # a new item has been added newlabel = request.form["newItem"] @@ -140,6 +154,7 @@ def items_append(listid): @items_bp.route("/items_multiappend/", methods=["GET", "POST"]) @login_required def items_multiappend(listid): + require_list_access(listid) if request.method == "POST": # a new item has been added newitems = request.form["newItems"] @@ -154,7 +169,7 @@ def items_multiappend(listid): @items_bp.route("/item_delete/") @login_required def item_delete(itemid): - i = Item.query.get(itemid) + i = require_item_access(itemid) listid = i.listofitems_id db.session.delete(i) db.session.commit() @@ -169,6 +184,7 @@ def item_delete(itemid): @items_bp.route("/items_clean/") @login_required def items_clean(listid): + require_list_access(listid) for i in Item.query.filter_by(listofitems_id=listid, is_checked=True, is_suggestion=False).all(): i.is_suggestion = True db.session.commit() @@ -178,7 +194,7 @@ def items_clean(listid): @items_bp.route("/item_addone/") @login_required def item_addone(itemid): - i = Item.query.get(itemid) + i = require_item_access(itemid) listid = i.listofitems_id # if it is checked or not on the list yet, set quantity to one if i.is_suggestion or i.is_checked: @@ -204,7 +220,7 @@ def item_addone(itemid): @items_bp.route("/item_addquantity//") @login_required def item_addquantity(itemid, quantity): - i = Item.query.get(itemid) + i = require_item_access(itemid) listid = i.listofitems_id # always add the item to the active todo if i.is_suggestion or i.is_checked: @@ -225,7 +241,7 @@ def item_addquantity(itemid, quantity): @items_bp.route("/item_update_category//") @login_required def item_upitem_update_categorydate(itemid, category): - i = Item.query.get(itemid) + i = require_item_access(itemid) i.category = category db.session.commit() - return '', 204 \ No newline at end of file + return '', 204 diff --git a/lists.py b/lists.py index 241af60..2a8d35f 100644 --- a/lists.py +++ b/lists.py @@ -1,4 +1,4 @@ -from flask import Blueprint, render_template, redirect, url_for, request, make_response +from flask import Blueprint, render_template, redirect, url_for, request, make_response, abort from flask_login import login_required, current_user from models import db from models import User, ListOfItems, Item, Shared @@ -32,17 +32,23 @@ def edit(): if request.method == "POST": if "iid" in request.form: ilist_id = int(request.form["iid"]) - newname = request.form["name"] + ilist = db.get_or_404(ListOfItems, ilist_id) + if not current_user.owns_list(ilist_id): + abort(403) + + newname = request.form["name"].strip() double = False for d in ListOfItems.query.filter_by(owner_user_id=current_user.id, name=newname).all(): if not d.id == ilist_id: # there is a duplicate name double = True if double: - return render_template("lists_edit.html", user=current_user, ilists=current_user.my_owned_lists(with_inactive=True), newlist=newlist, error_id=ilist_id) + return render_template("lists_edit.html", user=current_user, + ilists=current_user.my_owned_lists(with_inactive=True), + users=not_current_user(current_user.id), + newlist=newlist, error_id=ilist_id) # if id already exists, overwrite the existing list, else create a new one - ilist = ListOfItems.query.get(ilist_id) if ilist: # exists: delete OR overwrite shared data, activate data and update if "to_be_deleted" in request.form: @@ -68,12 +74,15 @@ def edit(): else: # must be a new list - newlist = ListOfItems(name=request.form["name"], owner_user_id=current_user.id) + newlist = ListOfItems(name=request.form["name"].strip(), owner_user_id=current_user.id) # check if name exists - duplicate = ListOfItems.query.filter_by(owner_user_id=current_user.id, name=request.form["name"]).first() + duplicate = ListOfItems.query.filter_by(owner_user_id=current_user.id, name=newlist.name).first() if duplicate: # there is a duplicate name - return render_template("lists_edit.html", user=current_user, ilists=current_user.my_owned_lists(with_inactive=True), newlist=newlist, error_id=9999) + return render_template("lists_edit.html", user=current_user, + ilists=current_user.my_owned_lists(with_inactive=True), + users=not_current_user(current_user.id), + newlist=newlist, error_id=9999) newlist.is_active = True newlist.owner_user_id = current_user.id diff --git a/models.py b/models.py index 52b0677..817d610 100644 --- a/models.py +++ b/models.py @@ -23,7 +23,6 @@ class User(UserMixin, db.Model): return result def set_password(self, password): - print(f"u:{self.name} > {password}") self.password_hash = generate_password_hash(password) def check_password(self, password): @@ -52,6 +51,19 @@ class User(UserMixin, db.Model): shared = Shared.query.filter_by(user_id=self.id, listofitem_id=listofitem_id).first() return True if shared else False + def owns_list(self, listofitem_id): + return ListOfItems.query.filter_by(id=listofitem_id, owner_user_id=self.id).first() is not None + + def can_access_list(self, listofitem_id): + listofitem = ListOfItems.query.filter_by(id=listofitem_id, is_active=True).first() + if not listofitem: + return False + return listofitem.owner_user_id == self.id or self.shares_in_list(listofitem_id) + + def can_access_item(self, item_id): + item = Item.query.get(item_id) + return item is not None and self.can_access_list(item.listofitems_id) + def set_group_id(self, secret): group = Group.query.filter_by(secret=secret).first() if group: @@ -72,7 +84,6 @@ class ListOfItems(db.Model): result[p] = getattr(self, p) result["pending"] = self.items_left() result["total"] = self.items_total() - print(result) return result def items_left(self): @@ -108,4 +119,3 @@ class Group(db.Model): id = db.Column(db.Integer, primary_key=True) updated_at = db.Column(db.TIMESTAMP, default=datetime.now(), onupdate=datetime.now(), nullable=False) secret = db.Column(db.String(128)) - diff --git a/templates/register.html b/templates/register.html index 78b7b1b..2d06ea1 100644 --- a/templates/register.html +++ b/templates/register.html @@ -1,6 +1,11 @@ {% extends "base.html" %} {% block content %}

+{% if error %} + +{% endif %}