summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--frest/auth/__init__.py0
-rw-r--r--frest/auth/__init__.py.bak0
-rw-r--r--frest/auth/forms.py10
-rw-r--r--frest/auth/models.py52
-rw-r--r--frest/auth/routes.py183
-rw-r--r--frest/decorators.py40
-rw-r--r--frest/utils.py3
7 files changed, 286 insertions, 2 deletions
diff --git a/frest/auth/__init__.py b/frest/auth/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/frest/auth/__init__.py
diff --git a/frest/auth/__init__.py.bak b/frest/auth/__init__.py.bak
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/frest/auth/__init__.py.bak
diff --git a/frest/auth/forms.py b/frest/auth/forms.py
new file mode 100644
index 0000000..abc2f49
--- /dev/null
+++ b/frest/auth/forms.py
@@ -0,0 +1,10 @@
+from .models import User
+from forms import ModelForm
+
+
+class UserForm(ModelForm):
+ model = User
+
+ def __init__(self, data):
+ super().__init__(self.model)
+ self.data = data
diff --git a/frest/auth/models.py b/frest/auth/models.py
new file mode 100644
index 0000000..ea79def
--- /dev/null
+++ b/frest/auth/models.py
@@ -0,0 +1,52 @@
+from database import db
+from datetime import datetime
+import string
+import random
+from hashlib import sha256
+from pytz import timezone
+import os
+
+
+def generate_token():
+ chars = string.ascii_uppercase + string.ascii_lowercase + string.digits
+ return "".join(random.choice(chars) for _ in range(18))
+
+
+class User(db.Model):
+ userId = db.Column(db.Integer, primary_key=True)
+ email = db.Column(db.String(30))
+ password = db.Column(db.String(30))
+ is_admin = db.Column(db.Boolean, default=False)
+ name = db.Column(db.String(30))
+ created_at = db.Column(db.DateTime)
+
+ def __init__(self, **kwargs):
+ self.email = kwargs.get("email")
+ psw_hash = sha256(kwargs.get("password").encode())
+ self.password = psw_hash.hexdigest()
+ self.name = kwargs.get("name")
+ self.is_admin = kwargs.get("is_admin")
+ self.created_at = datetime.now(
+ timezone(os.getenv("FREST_TIMEZONE", "Europe/Rome"))
+ )
+
+ def __repr__(self):
+ return f"<User '{self.userId}'>"
+
+
+class Token(db.Model):
+ tokenId = db.Column(db.Integer, primary_key=True)
+ string = db.Column(db.String(20))
+ created_at = db.Column(db.DateTime)
+ expired = db.Column(db.Boolean)
+ user_id = db.Column(db.Integer, db.ForeignKey("user.userId"), nullable=False)
+ user = db.relationship("User", backref=db.backref("tokens", lazy=True))
+
+ def __init__(self, user):
+ self.user = user
+ self.string = f"{generate_token()}=="
+ self.created_at = datetime.utcnow()
+ self.expired = False
+
+ def __repr__(self):
+ return f"<Token '{self.string}'>"
diff --git a/frest/auth/routes.py b/frest/auth/routes.py
new file mode 100644
index 0000000..c4dcfc9
--- /dev/null
+++ b/frest/auth/routes.py
@@ -0,0 +1,183 @@
+from flask import Blueprint, request, abort
+from utils import http_call, model_serialize
+from decorators import check_token, admin_required
+from .models import User, Token
+from .forms import UserForm
+from database import db
+from hashlib import sha256
+from sqlalchemy import desc
+
+api = Blueprint("users", __name__)
+
+
+@api.route("/api/login", methods=["POST"])
+def login():
+ if not request.json:
+ abort(400)
+
+ data = request.json
+
+ auth = request.headers.get("Authentication")
+ if auth:
+ t = Token.query.filter_by(string=auth).first()
+ if not t:
+ abort(404)
+
+ if t.user.is_admin:
+ return http_call(
+ {"userId": t.user.userId, "login": True, "token": t.string}, 200
+ )
+ else:
+ abort(403)
+
+ if "email" in data and "password" in data:
+ psw_hash = sha256(data["password"].encode())
+ data["password"] = psw_hash.hexdigest()
+ u = User.query.filter_by(email=data["email"], password=data["password"]).first()
+
+ if not u:
+ abort(404)
+
+ if "is_admin" in data:
+ if u.is_admin == 0:
+ abort(403)
+
+ last_token = (
+ Token.query.filter_by(user=u).order_by(desc(Token.tokenId)).all()[-1]
+ )
+ last_token.expired = True
+
+ t = Token(user=u)
+
+ db.session.add(t)
+ db.session.commit()
+
+ return http_call({"userId": u.userId, "login": True, "token": t.string}, 200)
+
+ abort(404)
+
+
+@api.route("/api/user/hash_password", methods=["GET"])
+def hash_password_exists():
+ data = request.args
+ if not data.get("hash_password"):
+ abort(400)
+
+ if User.query.filter_by(password=data["hash_password"]):
+ return http_call({}, 200)
+
+ return http_call({}, 404)
+
+
+@api.route("/api/user/new-password/<alias>", methods=["PUT"])
+def new_user_password(alias):
+ data = request.json
+ if not data.get("password"):
+ abort(400)
+
+ u = User.query.filter_by(password=alias).first()
+
+ if not u:
+ abort(404)
+
+ u.password = sha256(data["password"].encode()).hexdigest()
+ db.session.commit()
+
+ return http_call({}, 200)
+
+
+@api.route("/api/user", methods=["POST"])
+def new_user():
+ if not request.json:
+ abort(400)
+
+ form = UserForm(request.json)
+
+ if not form.get("is_admin") or form.is_valid():
+ if User.query.filter_by(email=form.get("email")).first():
+ abort(400)
+
+ u = User(
+ email=form.get("email"),
+ password=form.get("password"),
+ name=form.get("name"),
+ is_admin=form.get("is_admin"),
+ )
+ t = Token(user=u)
+ db.session.add(u)
+ db.session.add(t)
+
+ db.session.commit()
+
+ return http_call({"userId": u.userId, "token": t.string}, 201)
+
+ abort(400)
+
+
+@api.route("/api/users")
+@check_token
+@admin_required
+def all_users():
+ return http_call(
+ [
+ model_serialize(i, params="userId,email,is_admin,name,created_at")
+ for i in User.query.all()
+ ],
+ 200,
+ )
+
+
+@api.route("/api/user/<int:userId>")
+@check_token
+def get_user(userId):
+ return http_call(
+ model_serialize(
+ User.query.filter_by(userId=userId).first(),
+ params="userId,email,is_admin,name,created_at",
+ ),
+ 200,
+ )
+
+
+@api.route("/api/user/<userId>", methods=["DELETE"])
+@check_token
+def delete_user(userId):
+ u = User.query.filter_by(userId=userId)
+ if not u:
+ abort(404)
+
+ deleted = u.delete()
+ db.session.commit()
+
+ return http_call({"delete": deleted}, 200)
+
+
+@api.route("/api/user/<userId>", methods=["PUT"])
+@check_token
+def edit_user(userId):
+ if not request.json:
+ abort(400)
+
+ form = UserForm(request.json)
+ u = User.query.filter_by(userId=userId).first()
+ if not u:
+ abort(400)
+
+ if form.get("password"):
+ psw = True
+ else:
+ psw = False
+
+ if not psw or not form.get("is_admin") or form.is_valid():
+ u.name = form.get("name")
+ u.email = form.get("email")
+ u.is_admin = form.get("is_admin")
+ if psw:
+ crypt_psw = sha256(form.get("password").encode()).hexdigest()
+ u.password = crypt_psw
+
+ db.session.commit()
+
+ return http_call({"userId": u.userId}, 200)
+
+ abort(400)
diff --git a/frest/decorators.py b/frest/decorators.py
new file mode 100644
index 0000000..181b62d
--- /dev/null
+++ b/frest/decorators.py
@@ -0,0 +1,40 @@
+from flask import request, abort
+from auth.models import Token
+from functools import wraps
+
+
+def check_token(f):
+ @wraps(f)
+ def inner(*args, **kwargs):
+ userid = request.url.split('/')[-1]
+ headers = request.headers
+ if not headers.get("Authentication"):
+ abort(403)
+
+ auth = request.headers.get("Authentication")
+ token = Token.query.filter_by(string=auth).first()
+ if not token:
+ abort(403)
+
+ if userid.isdigit():
+ if int(userid) != token.user.userId and not token.user.is_admin:
+ abort(403)
+
+ return f(*args, **kwargs)
+
+ return inner
+
+
+def admin_required(f):
+ @wraps(f)
+ def inner(*args, **kwargs):
+ header = request.headers
+
+ auth = request.headers.get("Authentication")
+ token = Token.query.filter_by(string=auth).first()
+ if not token.user.is_admin:
+ abort(403)
+
+ return f(*args, **kwargs)
+
+ return inner
diff --git a/frest/utils.py b/frest/utils.py
index fbfa3bf..934aec4 100644
--- a/frest/utils.py
+++ b/frest/utils.py
@@ -1,7 +1,6 @@
-from flask import make_response, jsonify, abort, request, render_template
+from flask import make_response, jsonify, request, render_template
from flask_mail import Message
from mail import mail
-from functools import wraps
from datetime import datetime
import os
generated by cgit v1.2.3-18-g5258 (git 2.32.0) at 2024-11-22 10:28:53 +0000