group bugs and authorization improvements by codex

This commit is contained in:
2026-05-14 14:30:23 +02:00
parent a8554905ce
commit 0472ee1dd3
6 changed files with 98 additions and 50 deletions
+23 -24
View File
@@ -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
+16 -7
View File
@@ -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")
+23 -7
View File
@@ -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/<listid>")
@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/<itemid>/<itemchecked>")
@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/<listid>", 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/<listid>", 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/<itemid>")
@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/<listid>")
@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/<itemid>")
@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/<itemid>/<quantity>")
@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/<itemid>/<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
+16 -7
View File
@@ -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
+13 -3
View File
@@ -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))
+5
View File
@@ -1,6 +1,11 @@
{% extends "base.html" %}
{% block content %}
<h3 class="text-center mb-4"><img style="width: 3em; height: 3em" src="/static/register.svg"></h3>
{% if error %}
<div class="alert alert-danger" role="alert">
{{ error }}
</div>
{% endif %}
<form method="post">
<input class="form-control mb-3" name="name" placeholder="New Name" required>
<input class="form-control mb-3" name="password" type="password" placeholder="New Password" required>