mirror of
https://github.com/dancojocaru2000/foxbank.git
synced 2025-06-19 11:02:28 +03:00
Compare commits
5 commits
f91b6be3a5
...
02cf164620
Author | SHA1 | Date | |
---|---|---|---|
02cf164620 | |||
17d1cb2400 | |||
31511f6004 | |||
9ded9cc604 | |||
f19aad8d3e |
6 changed files with 98 additions and 111 deletions
|
@ -4,8 +4,14 @@ from flask_smorest import Api
|
||||||
from .accounts import bp as acc_bp
|
from .accounts import bp as acc_bp
|
||||||
from .login import bp as login_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):
|
def init_apis(app: Flask):
|
||||||
api = Api(app, spec_kwargs={
|
api = ApiWithErr(app, spec_kwargs={
|
||||||
'title': 'FoxBank',
|
'title': 'FoxBank',
|
||||||
'version': '1',
|
'version': '1',
|
||||||
'openapi_version': '3.0.0',
|
'openapi_version': '3.0.0',
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
from http import HTTPStatus
|
from http import HTTPStatus
|
||||||
|
from flask.views import MethodView
|
||||||
from flask_smorest import Blueprint, abort
|
from flask_smorest import Blueprint, abort
|
||||||
from marshmallow import Schema, fields
|
from marshmallow import Schema, fields
|
||||||
from ..decorators import ensure_logged_in
|
from ..decorators import ensure_logged_in
|
||||||
|
@ -23,60 +24,88 @@ class MetaAccountTypesSchema(Schema):
|
||||||
@bp.get('/meta/currencies')
|
@bp.get('/meta/currencies')
|
||||||
@bp.response(200, MetaCurrenciesSchema)
|
@bp.response(200, MetaCurrenciesSchema)
|
||||||
def get_valid_currencies():
|
def get_valid_currencies():
|
||||||
|
"""Get valid account currencies"""
|
||||||
return returns.success(currencies=VALID_CURRENCIES)
|
return returns.success(currencies=VALID_CURRENCIES)
|
||||||
|
|
||||||
|
|
||||||
@bp.get('/meta/account_types')
|
@bp.get('/meta/account_types')
|
||||||
@bp.response(200, MetaAccountTypesSchema)
|
@bp.response(200, MetaAccountTypesSchema)
|
||||||
def get_valid_account_types():
|
def get_valid_account_types():
|
||||||
|
"""Get valid account types"""
|
||||||
return returns.success(account_types=ACCOUNT_TYPES)
|
return returns.success(account_types=ACCOUNT_TYPES)
|
||||||
|
|
||||||
|
|
||||||
|
class AccountResponseSchema(returns.SuccessSchema):
|
||||||
|
account = fields.Nested(Account.Schema)
|
||||||
|
|
||||||
|
|
||||||
@bp.get('/<int:account_id>')
|
@bp.get('/<int:account_id>')
|
||||||
@ensure_logged_in
|
@ensure_logged_in
|
||||||
|
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
||||||
@bp.doc(security=[{'Token': []}])
|
@bp.doc(security=[{'Token': []}])
|
||||||
|
@bp.response(200, AccountResponseSchema)
|
||||||
def get_account_id(account_id: int):
|
def get_account_id(account_id: int):
|
||||||
|
"""Get account by id"""
|
||||||
account = db_utils.get_account(account_id=account_id)
|
account = db_utils.get_account(account_id=account_id)
|
||||||
if account is None:
|
if account is None:
|
||||||
return returns.NOT_FOUND
|
return returns.abort(returns.NOT_FOUND)
|
||||||
if decorators.user_id != db_utils.whose_account(account):
|
if decorators.user_id != db_utils.whose_account(account):
|
||||||
return returns.UNAUTHORIZED
|
return returns.abort(returns.UNAUTHORIZED)
|
||||||
account = account.to_json()
|
account = account.to_json()
|
||||||
return returns.success(account=account)
|
return returns.success(account=account)
|
||||||
|
|
||||||
|
|
||||||
@bp.get('/IBAN_<iban>')
|
@bp.get('/IBAN_<iban>')
|
||||||
@ensure_logged_in
|
@ensure_logged_in
|
||||||
|
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
||||||
@bp.doc(security=[{'Token': []}])
|
@bp.doc(security=[{'Token': []}])
|
||||||
|
@bp.response(200, AccountResponseSchema)
|
||||||
def get_account_iban(iban: str):
|
def get_account_iban(iban: str):
|
||||||
|
"""Get account by IBAN"""
|
||||||
account = db_utils.get_account(iban=iban)
|
account = db_utils.get_account(iban=iban)
|
||||||
if account is None:
|
if account is None:
|
||||||
return returns.NOT_FOUND
|
return returns.abort(returns.NOT_FOUND)
|
||||||
if decorators.user_id != db_utils.whose_account(account):
|
if decorators.user_id != db_utils.whose_account(account):
|
||||||
return returns.UNAUTHORIZED
|
return returns.abort(returns.UNAUTHORIZED)
|
||||||
account = account.to_json()
|
account = account.to_json()
|
||||||
return returns.success(account=account)
|
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')
|
||||||
|
|
||||||
class CreateAccountParams(Schema):
|
class CreateAccountResponseSchema(returns.SuccessSchema):
|
||||||
currency = fields.String()
|
account = fields.Nested(Account.Schema)
|
||||||
account_type = fields.String(data_key='accountType')
|
|
||||||
custom_name = fields.String(data_key='customName')
|
|
||||||
|
|
||||||
@bp.post('/')
|
@ensure_logged_in
|
||||||
@ensure_logged_in
|
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
||||||
@bp.arguments(CreateAccountParams, as_kwargs=True)
|
@bp.doc(security=[{'Token': []}])
|
||||||
@bp.response(200, Account.Schema)
|
@bp.arguments(CreateAccountParams, as_kwargs=True)
|
||||||
@bp.response(HTTPStatus.UNPROCESSABLE_ENTITY, description='Invalid currency or account type')
|
@bp.response(200, CreateAccountResponseSchema)
|
||||||
@bp.doc(security=[{'Token': []}])
|
@bp.response(HTTPStatus.UNPROCESSABLE_ENTITY, description='Invalid currency or account type')
|
||||||
def create_account(currency: str, account_type: str, custom_name: str):
|
def post(self, currency: str, account_type: str, custom_name: str):
|
||||||
if currency not in VALID_CURRENCIES:
|
"""Create account"""
|
||||||
abort(HTTPStatus.UNPROCESSABLE_ENTITY)
|
if currency not in VALID_CURRENCIES:
|
||||||
if account_type not in ACCOUNT_TYPES:
|
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'))
|
||||||
|
|
||||||
account = Account(-1, '', currency, account_type, custom_name or '')
|
account = Account(-1, '', currency, account_type, custom_name or '')
|
||||||
db_utils.insert_account(decorators.user_id, account)
|
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))
|
||||||
|
|
||||||
|
|
|
@ -26,17 +26,35 @@ class Login(MethodView):
|
||||||
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
@bp.response(401, returns.ErrorSchema, description='Login failure')
|
||||||
@bp.response(200, LoginSuccessSchema)
|
@bp.response(200, LoginSuccessSchema)
|
||||||
def post(self, username: str, code: str):
|
def post(self, username: str, code: str):
|
||||||
|
"""Login via username and TOTP code"""
|
||||||
user: User | None = get_user(username=username)
|
user: User | None = get_user(username=username)
|
||||||
if user is None:
|
if user is None:
|
||||||
return returns.INVALID_DETAILS
|
return returns.abort(returns.INVALID_DETAILS)
|
||||||
|
|
||||||
otp = TOTP(user.otp)
|
otp = TOTP(user.otp)
|
||||||
if not otp.verify(code, valid_window=1):
|
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)
|
token = ram_db.login_user(user.id)
|
||||||
return returns.success(token=token)
|
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')
|
@bp.route('/whoami')
|
||||||
class WhoAmI(MethodView):
|
class WhoAmI(MethodView):
|
||||||
class WhoAmISchema(returns.SuccessSchema):
|
class WhoAmISchema(returns.SuccessSchema):
|
||||||
|
@ -47,6 +65,7 @@ class WhoAmI(MethodView):
|
||||||
@bp.doc(security=[{'Token': []}])
|
@bp.doc(security=[{'Token': []}])
|
||||||
@ensure_logged_in
|
@ensure_logged_in
|
||||||
def get(self):
|
def get(self):
|
||||||
|
"""Get information about currently logged in user"""
|
||||||
user: User | None = get_user(user_id=decorators.user_id)
|
user: User | None = get_user(user_id=decorators.user_id)
|
||||||
if user is not None:
|
if user is not None:
|
||||||
user = user.to_json()
|
user = user.to_json()
|
||||||
|
|
|
@ -49,13 +49,13 @@ class Module(ModuleType):
|
||||||
def wrapper(*args, **kargs):
|
def wrapper(*args, **kargs):
|
||||||
token = request.headers.get('Authorization', None)
|
token = request.headers.get('Authorization', None)
|
||||||
if token is None:
|
if token is None:
|
||||||
return returns.NO_AUTHORIZATION
|
return returns.abort(returns.NO_AUTHORIZATION)
|
||||||
if not token.startswith('Bearer '):
|
if not token.startswith('Bearer '):
|
||||||
return returns.INVALID_AUTHORIZATION
|
return returns.abort(returns.INVALID_AUTHORIZATION)
|
||||||
token = token[7:]
|
token = token[7:]
|
||||||
user_id = ram_db.get_user(token)
|
user_id = ram_db.get_user(token)
|
||||||
if user_id is None:
|
if user_id is None:
|
||||||
return returns.INVALID_AUTHORIZATION
|
return returns.abort(returns.INVALID_AUTHORIZATION)
|
||||||
|
|
||||||
global _token
|
global _token
|
||||||
_token = token
|
_token = token
|
||||||
|
@ -71,40 +71,4 @@ class Module(ModuleType):
|
||||||
|
|
||||||
return wrapper
|
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__)
|
sys.modules[__name__] = Module(__name__)
|
||||||
|
|
|
@ -31,6 +31,13 @@ NOT_FOUND = _make_error(
|
||||||
'general/not_found',
|
'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
|
# Login
|
||||||
|
|
||||||
INVALID_DETAILS = _make_error(
|
INVALID_DETAILS = _make_error(
|
||||||
|
@ -76,3 +83,12 @@ class ErrorSchema(Schema):
|
||||||
|
|
||||||
class SuccessSchema(Schema):
|
class SuccessSchema(Schema):
|
||||||
status = fields.Constant('success')
|
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
|
||||||
|
|
|
@ -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)
|
|
||||||
|
|
Loading…
Add table
Reference in a new issue