server/foxbank_server/apis/__init__.py

@@ -4,14 +4,8 @@ from flask_smorest import Api
 from .accounts import bp as acc_bp
 from .login import bp as login_bp
 
-class ApiWithErr(Api):
-    def handle_http_exception(self, error):
-        if error.data and error.data['response']:
-            return error.data['response']
-        return super().handle_http_exception(error)
-
 def init_apis(app: Flask):
-    api = ApiWithErr(app, spec_kwargs={
+    api = Api(app, spec_kwargs={
         'title': 'FoxBank',
         'version': '1',
         'openapi_version': '3.0.0',

server/foxbank_server/apis/accounts.py

@@ -1,5 +1,4 @@
 from http import HTTPStatus
-from flask.views import MethodView
 from flask_smorest import Blueprint, abort
 from marshmallow import Schema, fields
 from ..decorators import ensure_logged_in
@@ -24,88 +23,60 @@ class MetaAccountTypesSchema(Schema):
 @bp.get('/meta/currencies')
 @bp.response(200, MetaCurrenciesSchema)
 def get_valid_currencies():
-    """Get valid account currencies"""
     return returns.success(currencies=VALID_CURRENCIES)
 
 
 @bp.get('/meta/account_types')
 @bp.response(200, MetaAccountTypesSchema)
 def get_valid_account_types():
-    """Get valid account types"""
     return returns.success(account_types=ACCOUNT_TYPES)
 
 
-class AccountResponseSchema(returns.SuccessSchema):
-    account = fields.Nested(Account.Schema)
-
-
 @bp.get('/<int:account_id>')
 @ensure_logged_in
-@bp.response(401, returns.ErrorSchema, description='Login failure')
 @bp.doc(security=[{'Token': []}])
-@bp.response(200, AccountResponseSchema)
 def get_account_id(account_id: int):
-    """Get account by id"""
     account = db_utils.get_account(account_id=account_id)
     if account is None:
-        return returns.abort(returns.NOT_FOUND)
+        return returns.NOT_FOUND
     if decorators.user_id != db_utils.whose_account(account):
-        return returns.abort(returns.UNAUTHORIZED)
+        return returns.UNAUTHORIZED
     account = account.to_json()
     return returns.success(account=account)
 
 
 @bp.get('/IBAN_<iban>')
 @ensure_logged_in
-@bp.response(401, returns.ErrorSchema, description='Login failure')
 @bp.doc(security=[{'Token': []}])
-@bp.response(200, AccountResponseSchema)
 def get_account_iban(iban: str):
-    """Get account by IBAN"""
     account = db_utils.get_account(iban=iban)
     if account is None:
-        return returns.abort(returns.NOT_FOUND)
+        return returns.NOT_FOUND
     if decorators.user_id != db_utils.whose_account(account):
-        return returns.abort(returns.UNAUTHORIZED)
+        return returns.UNAUTHORIZED
     account = account.to_json()
     return returns.success(account=account)
 
 
-@bp.route('/')
+
-class AccountsList(MethodView):
+class CreateAccountParams(Schema):
-    class CreateAccountParams(Schema):
     currency = fields.String()
     account_type = fields.String(data_key='accountType')
     custom_name = fields.String(data_key='customName')
 
-    class CreateAccountResponseSchema(returns.SuccessSchema):
+@bp.post('/')
-        account = fields.Nested(Account.Schema)
+@ensure_logged_in
-
+@bp.arguments(CreateAccountParams, as_kwargs=True)
-    @ensure_logged_in
+@bp.response(200, Account.Schema)
-    @bp.response(401, returns.ErrorSchema, description='Login failure')
+@bp.response(HTTPStatus.UNPROCESSABLE_ENTITY, description='Invalid currency or account type')
-    @bp.doc(security=[{'Token': []}])
+@bp.doc(security=[{'Token': []}])
-    @bp.arguments(CreateAccountParams, as_kwargs=True)
+def create_account(currency: str, account_type: str, custom_name: str):
-    @bp.response(200, CreateAccountResponseSchema)
-    @bp.response(HTTPStatus.UNPROCESSABLE_ENTITY, description='Invalid currency or account type')
-    def post(self, currency: str, account_type: str, custom_name: str):
-        """Create account"""
     if currency not in VALID_CURRENCIES:
-            return returns.abort(returns.invalid_argument('currency'))
+        abort(HTTPStatus.UNPROCESSABLE_ENTITY)
     if account_type not in ACCOUNT_TYPES:
-            return returns.abort(returns.invalid_argument('account_type'))
+        abort(HTTPStatus.UNPROCESSABLE_ENTITY)
 
     account = Account(-1, '', currency, account_type, custom_name or '')
     db_utils.insert_account(decorators.user_id, account)
-        return returns.success(account=account.to_json())
+    return account.to_json()
-
-    class AccountsResponseSchema(returns.SuccessSchema):
-        accounts = fields.List(fields.Nested(Account.Schema))
-
-    @ensure_logged_in
-    @bp.response(401, returns.ErrorSchema, description='Login failure')
-    @bp.doc(security=[{'Token': []}])
-    @bp.response(200, AccountsResponseSchema)
-    def get(self):
-        """Get all accounts of user"""
-        return returns.success(accounts=db_utils.get_accounts(decorators.user_id))
 

server/foxbank_server/apis/login.py

@@ -26,35 +26,17 @@ class Login(MethodView):
     @bp.response(401, returns.ErrorSchema, description='Login failure')
     @bp.response(200, LoginSuccessSchema)
     def post(self, username: str, code: str):
-        """Login via username and TOTP code"""
         user: User | None = get_user(username=username)
         if user is None:
-            return returns.abort(returns.INVALID_DETAILS)
+            return returns.INVALID_DETAILS
 
         otp = TOTP(user.otp)
         if not otp.verify(code, valid_window=1):
-            return returns.abort(returns.INVALID_DETAILS)
+            return returns.INVALID_DETAILS
 
         token = ram_db.login_user(user.id)
         return returns.success(token=token)
 
-    @ensure_logged_in
-    @bp.doc(security=[{'Token': []}])
-    @bp.response(401, returns.ErrorSchema, description='Login failure')
-    @bp.response(204)
-    def delete(self):
-        """Logout"""
-        ram_db.logout_user(decorators.token)
-
-@bp.post('/logout')
-@ensure_logged_in
-@bp.doc(security=[{'Token': []}])
-@bp.response(401, returns.ErrorSchema, description='Login failure')
-@bp.response(204)
-def logout_route():
-    """Logout"""
-    ram_db.logout_user(decorators.token)
-
 @bp.route('/whoami')
 class WhoAmI(MethodView):
     class WhoAmISchema(returns.SuccessSchema):
@@ -65,7 +47,6 @@ class WhoAmI(MethodView):
     @bp.doc(security=[{'Token': []}])
     @ensure_logged_in
     def get(self):
-        """Get information about currently logged in user"""
         user: User | None = get_user(user_id=decorators.user_id)
         if user is not None:
             user = user.to_json()

server/foxbank_server/decorators.py

@@ -49,13 +49,13 @@ class Module(ModuleType):
         def wrapper(*args, **kargs):
             token = request.headers.get('Authorization', None)
             if token is None:
-                return returns.abort(returns.NO_AUTHORIZATION)
+                return returns.NO_AUTHORIZATION
             if not token.startswith('Bearer '):
-                return returns.abort(returns.INVALID_AUTHORIZATION)
+                return returns.INVALID_AUTHORIZATION
             token = token[7:]
             user_id = ram_db.get_user(token)
             if user_id is None:
-                return returns.abort(returns.INVALID_AUTHORIZATION)
+                return returns.INVALID_AUTHORIZATION
 
             global _token
             _token = token
@@ -71,4 +71,40 @@ class Module(ModuleType):
 
         return wrapper
 
+    # def ensure_logged_in(token=False, user_id=False):
+    #     """
+    #     Ensure the user is logged in by providing an Authorization: Bearer token
+    #     header.
+    #
+    #     @param token whether the token should be supplied after validation
+    #     @param user_id whether the user_id should be supplied after validation
+    #     @return decorator which supplies the requested parameters
+    #     """
+    #     def decorator(fn):
+    #         pass_token = token
+    #         pass_user_id = user_id
+    #
+    #         @wraps(fn)
+    #         def wrapper(*args, **kargs):
+    #             token = request.headers.get('Authorization', None)
+    #             if token is None:
+    #                 return returns.NO_AUTHORIZATION
+    #             if not token.startswith('Bearer '):
+    #                 return returns.INVALID_AUTHORIZATION
+    #             token = token[7:]
+    #             user_id = ram_db.get_user(token)
+    #             if user_id is None:
+    #                 return returns.INVALID_AUTHORIZATION
+    #
+    #             if pass_user_id and pass_token:
+    #                 return fn(user_id=user_id, token=token, *args, **kargs)
+    #             elif pass_user_id:
+    #                 return fn(user_id=user_id, *args, **kargs)
+    #             elif pass_token:
+    #                 return fn(token=token, *args, **kargs)
+    #             else:
+    #                 return fn(*args, **kargs)
+    #         return wrapper
+    #     return decorator
+
 sys.modules[__name__] = Module(__name__)

server/foxbank_server/returns.py

@@ -31,13 +31,6 @@ NOT_FOUND = _make_error(
     'general/not_found',
 )
 
-def invalid_argument(argname: str) -> tuple[Any, int]:
-    return _make_error(
-        _HTTPStatus.UNPROCESSABLE_ENTITY,
-        'general/invalid_argument',
-        message=f'Invalid argument: {argname}',
-    )
-
 # Login
 
 INVALID_DETAILS = _make_error(
@@ -83,12 +76,3 @@ class ErrorSchema(Schema):
 
 class SuccessSchema(Schema):
     status = fields.Constant('success')
-
-# smorest
-
-def abort(result: tuple[Any, int]):
-    try:
-        from flask_smorest import abort as _abort
-        _abort(result[1], response=result)
-    except ImportError:
-        return result

server/login.py

@@ -0,0 +1,47 @@
+from functools import wraps
+from flask import Blueprint, request
+
+from pyotp import TOTP
+
+import db_utils
+from decorators import no_content, ensure_logged_in, user_id, token
+import models
+import ram_db
+import returns
+
+login = Blueprint('login', __name__)
+
+@login.post('/')
+def make_login():
+    try:
+        username = request.json['username']
+        code = request.json['code']
+    except (TypeError, KeyError):
+        return returns.INVALID_REQUEST
+
+    user: models.User | None = db_utils.get_user(username=username)
+    if user is None:
+        return returns.INVALID_DETAILS
+
+    otp = TOTP(user.otp)
+    if not otp.verify(code, valid_window=1):
+        return returns.INVALID_DETAILS
+
+    token = ram_db.login_user(user.id)
+    return returns.success(token=token)
+
+@login.post('/logout')
+@ensure_logged_in
+@no_content
+def logout():
+    ram_db.logout_user(token)
+
+@login.get('/whoami')
+@ensure_logged_in
+def whoami():
+    user: models.User | None = db_utils.get_user(user_id=user_id)
+    if user is not None:
+        user = user.to_json()
+
+    return returns.successs(user=user)
+