stable/freqtrade/rpc/api_server/api_auth.py

130 lines
4.6 KiB
Python
Raw Normal View History

import logging
2020-12-25 14:50:19 +00:00
import secrets
2020-12-26 14:54:22 +00:00
from datetime import datetime, timedelta
from typing import Any, Dict, Union
2020-12-25 14:50:19 +00:00
2020-12-26 14:54:22 +00:00
import jwt
from fastapi import APIRouter, Depends, HTTPException, WebSocket, status
2020-12-26 14:54:22 +00:00
from fastapi.security import OAuth2PasswordBearer
2020-12-25 14:50:19 +00:00
from fastapi.security.http import HTTPBasic, HTTPBasicCredentials
2020-12-26 14:54:22 +00:00
2021-01-02 08:07:31 +00:00
from freqtrade.rpc.api_server.api_schemas import AccessAndRefreshToken, AccessToken
2020-12-31 10:01:50 +00:00
from freqtrade.rpc.api_server.deps import get_api_config
2020-12-25 14:50:19 +00:00
logger = logging.getLogger(__name__)
2020-12-25 14:50:19 +00:00
ALGORITHM = "HS256"
router_login = APIRouter()
2020-12-27 14:24:49 +00:00
def verify_auth(api_config, username: str, password: str):
2020-12-26 07:48:15 +00:00
"""Verify username/password"""
2020-12-27 14:24:49 +00:00
return (secrets.compare_digest(username, api_config.get('username')) and
secrets.compare_digest(password, api_config.get('password')))
2020-12-25 14:50:19 +00:00
2020-12-26 07:48:15 +00:00
httpbasic = HTTPBasic(auto_error=False)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
2020-12-25 14:50:19 +00:00
2020-12-27 14:24:49 +00:00
def get_user_from_token(token, secret_key: str, token_type: str = "access"):
2020-12-26 07:48:15 +00:00
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
2020-12-27 14:24:49 +00:00
payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM])
username: str = payload.get("identity", {}).get('u')
2020-12-26 07:48:15 +00:00
if username is None:
raise credentials_exception
if payload.get("type") != token_type:
raise credentials_exception
2020-12-25 14:50:19 +00:00
2020-12-26 07:48:15 +00:00
except jwt.PyJWTError:
raise credentials_exception
return username
2020-12-25 14:50:19 +00:00
# This should be reimplemented to better realign with the existing tools provided
# by FastAPI regarding API Tokens
# https://github.com/tiangolo/fastapi/blob/master/fastapi/security/api_key.py
async def get_ws_token(
ws: WebSocket,
token: Union[str, None] = None,
api_config: Dict[str, Any] = Depends(get_api_config)
):
secret_ws_token = api_config['ws_token']
if token == secret_ws_token:
# Just return the token if it matches
return token
else:
logger.debug("Denying websocket request")
# If it doesn't match, close the websocket connection
await ws.close(code=status.WS_1008_POLICY_VIOLATION)
2021-01-08 18:27:51 +00:00
def create_token(data: dict, secret_key: str, token_type: str = "access") -> str:
2020-12-26 07:48:15 +00:00
to_encode = data.copy()
if token_type == "access":
expire = datetime.utcnow() + timedelta(minutes=15)
elif token_type == "refresh":
expire = datetime.utcnow() + timedelta(days=30)
else:
raise ValueError()
to_encode.update({
"exp": expire,
"iat": datetime.utcnow(),
"type": token_type,
})
2020-12-27 14:24:49 +00:00
encoded_jwt = jwt.encode(to_encode, secret_key, algorithm=ALGORITHM)
2020-12-26 07:48:15 +00:00
return encoded_jwt
def http_basic_or_jwt_token(form_data: HTTPBasicCredentials = Depends(httpbasic),
2020-12-27 14:24:49 +00:00
token: str = Depends(oauth2_scheme),
api_config=Depends(get_api_config)):
2020-12-26 07:48:15 +00:00
if token:
2020-12-27 14:24:49 +00:00
return get_user_from_token(token, api_config.get('jwt_secret_key', 'super-secret'))
elif form_data and verify_auth(api_config, form_data.username, form_data.password):
2020-12-26 07:48:15 +00:00
return form_data.username
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
2020-12-27 13:57:01 +00:00
detail="Unauthorized",
2020-12-26 07:48:15 +00:00
)
2020-12-25 14:50:19 +00:00
@router_login.post('/token/login', response_model=AccessAndRefreshToken)
2020-12-27 14:24:49 +00:00
def token_login(form_data: HTTPBasicCredentials = Depends(HTTPBasic()),
api_config=Depends(get_api_config)):
2020-12-25 14:50:19 +00:00
2020-12-27 14:24:49 +00:00
if verify_auth(api_config, form_data.username, form_data.password):
token_data = {'identity': {'u': form_data.username}}
2020-12-27 14:24:49 +00:00
access_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'))
refresh_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'),
token_type="refresh")
2020-12-25 14:50:19 +00:00
return {
"access_token": access_token,
"refresh_token": refresh_token,
}
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
@router_login.post('/token/refresh', response_model=AccessToken)
2020-12-27 14:24:49 +00:00
def token_refresh(token: str = Depends(oauth2_scheme), api_config=Depends(get_api_config)):
2020-12-26 07:48:15 +00:00
# Refresh token
2020-12-27 14:24:49 +00:00
u = get_user_from_token(token, api_config.get(
'jwt_secret_key', 'super-secret'), 'refresh')
token_data = {'identity': {'u': u}}
2020-12-27 14:24:49 +00:00
access_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'),
token_type="access")
2020-12-25 14:50:19 +00:00
return {'access_token': access_token}