1
0
Fork 0
mirror of https://github.com/dancojocaru2000/foxbank.git synced 2025-06-19 11:02:28 +03:00

Compare commits

..

5 commits

6 changed files with 98 additions and 111 deletions

View file

@ -4,8 +4,14 @@ 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 = Api(app, spec_kwargs={
api = ApiWithErr(app, spec_kwargs={
'title': 'FoxBank',
'version': '1',
'openapi_version': '3.0.0',

View file

@ -1,4 +1,5 @@
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
@ -23,60 +24,88 @@ 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.NOT_FOUND
return returns.abort(returns.NOT_FOUND)
if decorators.user_id != db_utils.whose_account(account):
return returns.UNAUTHORIZED
return returns.abort(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.NOT_FOUND
return returns.abort(returns.NOT_FOUND)
if decorators.user_id != db_utils.whose_account(account):
return returns.UNAUTHORIZED
return returns.abort(returns.UNAUTHORIZED)
account = account.to_json()
return returns.success(account=account)
@bp.route('/')
class AccountsList(MethodView):
class CreateAccountParams(Schema):
currency = fields.String()
account_type = fields.String(data_key='accountType')
custom_name = fields.String(data_key='customName')
@bp.post('/')
class CreateAccountResponseSchema(returns.SuccessSchema):
account = fields.Nested(Account.Schema)
@ensure_logged_in
@bp.arguments(CreateAccountParams, as_kwargs=True)
@bp.response(200, Account.Schema)
@bp.response(HTTPStatus.UNPROCESSABLE_ENTITY, description='Invalid currency or account type')
@bp.response(401, returns.ErrorSchema, description='Login failure')
@bp.doc(security=[{'Token': []}])
def create_account(currency: str, account_type: str, custom_name: str):
@bp.arguments(CreateAccountParams, as_kwargs=True)
@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:
abort(HTTPStatus.UNPROCESSABLE_ENTITY)
return returns.abort(returns.invalid_argument('currency'))
if account_type not in ACCOUNT_TYPES:
abort(HTTPStatus.UNPROCESSABLE_ENTITY)
return returns.abort(returns.invalid_argument('account_type'))
account = Account(-1, '', currency, account_type, custom_name or '')
db_utils.insert_account(decorators.user_id, account)
return account.to_json()
return returns.success(account=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))

View file

@ -26,17 +26,35 @@ 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.INVALID_DETAILS
return returns.abort(returns.INVALID_DETAILS)
otp = TOTP(user.otp)
if not otp.verify(code, valid_window=1):
return returns.INVALID_DETAILS
return returns.abort(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):
@ -47,6 +65,7 @@ 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()

View file

@ -49,13 +49,13 @@ class Module(ModuleType):
def wrapper(*args, **kargs):
token = request.headers.get('Authorization', None)
if token is None:
return returns.NO_AUTHORIZATION
return returns.abort(returns.NO_AUTHORIZATION)
if not token.startswith('Bearer '):
return returns.INVALID_AUTHORIZATION
return returns.abort(returns.INVALID_AUTHORIZATION)
token = token[7:]
user_id = ram_db.get_user(token)
if user_id is None:
return returns.INVALID_AUTHORIZATION
return returns.abort(returns.INVALID_AUTHORIZATION)
global _token
_token = token
@ -71,40 +71,4 @@ 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__)

View file

@ -31,6 +31,13 @@ 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(
@ -76,3 +83,12 @@ 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

View file

@ -1,47 +0,0 @@
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)