Changing basic things like auth for api and laying out basic structure.

.python-version

@@ -0,0 +1 @@
+3.11

dev.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
 PORT="${PORT:-7080}"
-export FLASK_APP=tpnewsletter
+export FLASK_APP=tread
 export FLASK_ENV=development
-flask run --port $PORT --host 0.0.0.0 --debug
+flask run --port $PORT --host 0.0.0.0 --debug

pyproject.toml

@@ -10,6 +10,8 @@ license = { file = "LICENSE" }
 dependencies = [
  "flask",
  "flask-bcrypt",
+ "flask-cors",
+ "flask-jwt-extended[asymmetric_crypto]",
  "flask-sqlalchemy",
  "flask-wtf",
  "flask-migrate",

requirements.txt

@@ -7,6 +7,7 @@ passlib[bcrypt]==1.7.4
 
 requests==2.32.3
 
+flask-jwt-extended[asymmetric_crypto]
 flask-sqlalchemy
 flask-bcrypt
 flask-mail

tread/__init__.py

@@ -2,7 +2,8 @@
 from pathlib import Path
 from flask import Flask
 from .config import Config
-from .routes.frontend import frontend
+from .jwt import jwt
+from .routes.webui import webui
 from .routes.api import api
 from .bcrypt import bcrypt
 from .database import db
@@ -18,7 +19,6 @@ def create_app(test_config=None):
  # load the test config if passed in
  app.config.from_mapping(test_config)
 
- print(str(app.config))
  try: 
  Path(app.config['DATA_DIR']).mkdir(parents=True, exist_ok=True)
  except OSError:
@@ -49,17 +49,16 @@ def create_app(test_config=None):
  except OSError:
  pass
 
- app.register_blueprint(frontend)
+ app.register_blueprint(webui)
  app.register_blueprint(api)
 
  bcrypt.init_app(app)
+ jwt.init_app(app)
  db.init_app(app)
 
  with app.app_context():
  db.create_all()
 
- TOKEN_EXPIRATION_MINUTES = app.config['TOKEN_EXPIRATION_MINUTES']
-
  return app
 
 

tread/config.py

@@ -4,7 +4,9 @@
 import random
 from pathlib import Path
 from unittest.mock import DEFAULT
+import logging
 
+log = logging.getLogger(__name__)
 class Config:
 
  if platform.system() == 'Darwin' or platform.system() == 'Windows':
@@ -25,32 +27,43 @@ class Config:
  BUILD_DIR: str = str(Path(os.environ.get("BUILD_DIR", default=APP_ROOT + "/frontend/public")))
 
  KEY_FILE: str = APP_DIR + "/.secret_key"
-
- if os.environ.get("SECRET_KEY") is None:
- if not KEY_FILE.exists():
- with open(KEY_FILE, "wb") as f:
- f.write(base64.b64encode(os.urandom(32)).decode("utf-8"))
- f.close()
-
- with open(KEY_FILE, "r") as f:
- SECRET_KEY = f.read()
- f.close()
-
- SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL", default="sqlite:///" + DB_DIR + "/tpnewsletter.db")
 
- JWT_COOKIE_NAME = os.environ.get("JWT_COOKIE_NAME", default="jwt_token")
- JWT_COOKIE_SECURE = bool(os.environ.get("JWT_COOKIE_SECURE", default=False))
- JWT_COOKIE_HTTPONLY = bool(os.environ.get("JWT_COOKIE_HTTPONLY", default=True))
+ SECRET_KEY = os.environ.get("SECRET_KEY", default=None)
+
+ SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL", default="sqlite:///" + DB_DIR + "/tread.db")
+
+ JWT_SECRET_KEY = SECRET_KEY
 
- TOKEN_EXPIRATION_MINUTES = int(os.environ.get("TOKEN_EXPIRATION_MINUTES", default=60))
+ JWT_COOKIE_NAME = "jwt_tread"
+ JWT_COOKIE_SECURE = os.environ.get("JWT_COOKIE_SECURE", default=False)
+ JWT_COOKIE_HTTPONLY = os.environ.get("JWT_COOKIE_HTTPONLY", default=True)
+
+ # in hours but converted to minutes for more accurate token expiration
+ JWT_TOKEN_EXPIRATION = os.environ.get("JWT_TOKEN_EXPIRATION", default=(24 * 60))
 
  ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", default="admin")
  ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", default="P@assw0rd")
+ UPLOAD_MAX_SIZE = int(os.environ.get("UPLOAD_MAX_SIZE", default=1000000000))
 
- DEFAULT_USER_ROLE = os.environ.get("DEFAULT_USER_ROLE", default="user")
- DEFAULT_USER_THEME = os.environ.get("DEFAULT_USER_THEME", default="light")
- DEFAULT_USER_PROFILE_PICTURE_URL = os.environ.get("DEFAULT_USER_PROFILE_PICTURE_URL", default="https://i.imgur.com/6kOZ4H1.png")
-
+ def __init__(self):
+ if self.SECRET_KEY==None:
+ self.SECRET_KEY: None = Config.get_secret_key()
+
+ log.info(msg="SECRET_KEY: " + self.SECRET_KEY)
 
+ @staticmethod
+ def get_secret_key():
+ if os.environ.get("SECRET_KEY") is None:
+ if not Config.KEY_FILE.exists():
+ with open(Config.KEY_FILE, "wb") as f:
+ f.write(base64.b64encode(os.urandom(32)).decode("utf-8"))
+ f.close()
+ log.info(msg="Secret key created.")
+
+ with open(Config.KEY_FILE, "r") as f:
+ secret_key = f.read()
+ f.close()
 
+ return secret_key
+
 

tread/database/user.py

@@ -1,18 +1,42 @@
 from typing import Any
 from . import db
 from ..bcrypt import bcrypt
+import logging
+
+log = logging.getLogger(__name__)
+log.setLevel(logging.WARN)
 
 class User(db.Model):
  id = db.Column(db.Integer, primary_key=True)
  username = db.Column(db.String(150), unique=True, nullable=False)
- password_hash: Any = db.Column(db.String(150), nullable=False)
+ password: Any = db.Column(db.String(150), nullable=False)
+ full_name = db.Column(db.String(150), nullable=False)
  role = db.Column(db.String(50), nullable=False)
- theme: Any = db.Column(db.String(50), nullable=True)
- profile_picture_url: Any = db.Column(db.String(250), nullable=True)
+ email = db.Column(db.String(150), nullable=False)
  created_at = db.Column(db.DateTime, nullable=False, default=db.func.now())
+ updated_at = db.Column(db.DateTime, nullable=False, default=db.func.now(), onupdate=db.func.now())
 
+ def __init__(self, username, password, full_name, role, email):
+ self.username = username
+ self.set_password(password) # hash the password
+ self.full_name = full_name
+ self.role = role
+ self.email = email
+
  def set_password(self, password):
- self.password_hash = bcrypt.generate_password_hash(password).decode("utf-8")
+ self.password = bcrypt.generate_password_hash(password).decode("utf-8")
 
  def check_password(self, password):
- return bcrypt.check_password_hash(self.password_hash, password)
+ return bcrypt.check_password_hash(self.password_hash, password)
+
+ def save(self):
+ try:
+ db.session.add(self)
+ db.session.commit()
+ log.info(msg="User saved to database.")
+ except:
+ db.session.rollback()
+ log.error(msg="Error saving user to database, rolling back.")
+ return False
+
+ return True

tread/jwt/__init__.py

@@ -1,38 +1,19 @@
-from typing import Any
-from flask import request, current_app, jsonify
-from functools import wraps
-import logging
-import jwt
-import datetime
-
-log = logging.getLogger(__name__)
-log.setLevel(logging.WARN)
-
-# Decorator to check the JWT in cookies
-def token_required(f):
- @wraps(f)
- def decorated(*args, **kwargs):
- token = request.cookies.get(current_app.config['JWT_COOKIE_NAME'])
-
- if not token:
- log.info(msg="Token is missing. Returning 403.")
- return jsonify({'message': 'Token is missing'}), 403
-
- try:
- # Decode the token and extract user information
- data = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256'])
- current_user = data['user']
- role: str = data['role']
- created_at: Any = data['created_at']
- exp: datetime.datetime = data['exp']
- except:
- log.info(msg='Token is invalid. Returning 403.')
- return jsonify({'message': 'Token is invalid'}), 403
-
- if exp < datetime.datetime.now(datetime.timezone.utc):
- log.info(msg='Token has expired. Returning 403.')
- return jsonify({'message': 'Token has expired'}), 403
-
- return f(current_user, role, created_at, exp, *args, **kwargs)
-
- return decorated
+from flask_jwt_extended import JWTManager
+from flask_jwt_extended import current_user
+from ..database.user import User
+
+jwt = JWTManager()
+
+@jwt.user_identity_loader
+def user_identity_lookup(user):
+ return user.id
+
+@jwt.user_lookup_loader
+def user_lookup_callback(_jwt_header, jwt_data):
+ identity = jwt_data["sub"]
+ return User.query.filter_by(id=identity).one_or_none()
+
+
+
+
+