9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.4

/freqtrade/rpc/api_server/api_auth.py

import logging
import secrets
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Union

import jwt
from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.security.http import HTTPBasic, HTTPBasicCredentials

from freqtrade.rpc.api_server.api_schemas import AccessAndRefreshToken, AccessToken
from freqtrade.rpc.api_server.deps import get_api_config


logger = logging.getLogger(__name__)

ALGORITHM = "HS256"

router_login = APIRouter()


def verify_auth(api_config, username: str, password: str):
    """Verify username/password"""
    return (secrets.compare_digest(username, api_config.get('username')) and
            secrets.compare_digest(password, api_config.get('password')))


httpbasic = HTTPBasic(auto_error=False)
security = HTTPBasic()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)


def get_user_from_token(token, secret_key: str, token_type: str = "access") -> str:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM])
        username: str = payload.get("identity", {}).get('u')
        if username is None:
            raise credentials_exception
        if payload.get("type") != token_type:
            raise credentials_exception

    except jwt.PyJWTError:
        raise credentials_exception
    return username


# 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 validate_ws_token(
    ws: WebSocket,
    ws_token: Union[str, None] = Query(default=None, alias="token"),
    api_config: Dict[str, Any] = Depends(get_api_config)
):
    secret_ws_token = api_config.get('ws_token', None)
    secret_jwt_key = api_config.get('jwt_secret_key', 'super-secret')

    # Check if ws_token is/in secret_ws_token
    if ws_token and secret_ws_token:
        is_valid_ws_token = False
        if isinstance(secret_ws_token, str):
            is_valid_ws_token = secrets.compare_digest(secret_ws_token, ws_token)
        elif isinstance(secret_ws_token, list):
            is_valid_ws_token = any([
                secrets.compare_digest(potential, ws_token)
                for potential in secret_ws_token
            ])

        if is_valid_ws_token:
            return ws_token

    # Check if ws_token is a JWT
    try:
        user = get_user_from_token(ws_token, secret_jwt_key)
        return user
    # If the token is a jwt, and it's valid return the user
    except HTTPException:
        pass

    # If it doesn't match, close the websocket connection
    await ws.close(code=status.WS_1008_POLICY_VIOLATION)


def create_token(data: dict, secret_key: str, token_type: str = "access") -> str:
    to_encode = data.copy()
    if token_type == "access":
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    elif token_type == "refresh":
        expire = datetime.now(timezone.utc) + timedelta(days=30)
    else:
        raise ValueError()
    to_encode.update({
        "exp": expire,
        "iat": datetime.now(timezone.utc),
        "type": token_type,
    })
    encoded_jwt = jwt.encode(to_encode, secret_key, algorithm=ALGORITHM)
    return encoded_jwt


def http_basic_or_jwt_token(form_data: HTTPBasicCredentials = Depends(httpbasic),
                            token: str = Depends(oauth2_scheme),
                            api_config=Depends(get_api_config)):
    if token:
        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):
        return form_data.username

    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Unauthorized",
    )


@router_login.post('/token/login', response_model=AccessAndRefreshToken)
def token_login(form_data: HTTPBasicCredentials = Depends(security),
                api_config=Depends(get_api_config)):

    if verify_auth(api_config, form_data.username, form_data.password):
        token_data = {'identity': {'u': form_data.username}}
        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")
        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)
def token_refresh(token: str = Depends(oauth2_scheme), api_config=Depends(get_api_config)):
    # Refresh token
    u = get_user_from_token(token, api_config.get(
        'jwt_secret_key', 'super-secret'), 'refresh')
    token_data = {'identity': {'u': u}}
    access_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'),
                                token_type="access")
    return {'access_token': access_token}

1	import logging	1✔
2	import secrets	1✔
3	from datetime import datetime, timedelta, timezone	1✔
4	from typing import Any, Dict, Union	1✔
5
6	import jwt	1✔
7	from fastapi import APIRouter, Depends, HTTPException, Query, WebSocket, status	1✔
8	from fastapi.security import OAuth2PasswordBearer	1✔
9	from fastapi.security.http import HTTPBasic, HTTPBasicCredentials	1✔
10
11	from freqtrade.rpc.api_server.api_schemas import AccessAndRefreshToken, AccessToken	1✔
12	from freqtrade.rpc.api_server.deps import get_api_config	1✔
13
14
15	logger = logging.getLogger(__name__)	1✔
16
17	ALGORITHM = "HS256"	1✔
18
19	router_login = APIRouter()	1✔
20
21
22	def verify_auth(api_config, username: str, password: str):	1✔
23	"""Verify username/password"""
24	return (secrets.compare_digest(username, api_config.get('username')) and	1✔
25	secrets.compare_digest(password, api_config.get('password')))
26
27
28	httpbasic = HTTPBasic(auto_error=False)	1✔
29	security = HTTPBasic()	1✔
30	oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)	1✔
31
32
33	def get_user_from_token(token, secret_key: str, token_type: str = "access") -> str:	1✔
34	credentials_exception = HTTPException(	1✔
35	status_code=status.HTTP_401_UNAUTHORIZED,
36	detail="Could not validate credentials",
37	headers={"WWW-Authenticate": "Bearer"},
38	)
39	try:	1✔
40	payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM])	1✔
41	username: str = payload.get("identity", {}).get('u')	1✔
42	if username is None:	1✔
43	raise credentials_exception	1✔
44	if payload.get("type") != token_type:	1✔
45	raise credentials_exception	1✔
46
47	except jwt.PyJWTError:	1✔
48	raise credentials_exception	1✔
49	return username	1✔
50
51
52	# This should be reimplemented to better realign with the existing tools provided
53	# by FastAPI regarding API Tokens
54	# https://github.com/tiangolo/fastapi/blob/master/fastapi/security/api_key.py
55	async def validate_ws_token(	1✔
56	ws: WebSocket,
57	ws_token: Union[str, None] = Query(default=None, alias="token"),
58	api_config: Dict[str, Any] = Depends(get_api_config)
59	):
60	secret_ws_token = api_config.get('ws_token', None)	1✔
61	secret_jwt_key = api_config.get('jwt_secret_key', 'super-secret')	1✔
62
63	# Check if ws_token is/in secret_ws_token
64	if ws_token and secret_ws_token:	1✔
65	is_valid_ws_token = False	1✔
66	if isinstance(secret_ws_token, str):	1✔
67	is_valid_ws_token = secrets.compare_digest(secret_ws_token, ws_token)	1✔
68	elif isinstance(secret_ws_token, list):	×
69	is_valid_ws_token = any([	×
70	secrets.compare_digest(potential, ws_token)
71	for potential in secret_ws_token
72	])
73
74	if is_valid_ws_token:	1✔
75	return ws_token	1✔
76
77	# Check if ws_token is a JWT
78	try:	1✔
79	user = get_user_from_token(ws_token, secret_jwt_key)	1✔
80	return user	1✔
81	# If the token is a jwt, and it's valid return the user
82	except HTTPException:	1✔
83	pass	1✔
84
85	# If it doesn't match, close the websocket connection
86	await ws.close(code=status.WS_1008_POLICY_VIOLATION)	1✔
87
88
89	def create_token(data: dict, secret_key: str, token_type: str = "access") -> str:	1✔
90	to_encode = data.copy()	1✔
91	if token_type == "access":	1✔
92	expire = datetime.now(timezone.utc) + timedelta(minutes=15)	1✔
93	elif token_type == "refresh":	1✔
94	expire = datetime.now(timezone.utc) + timedelta(days=30)	1✔
95	else:
96	raise ValueError()	1✔
97	to_encode.update({	1✔
98	"exp": expire,
99	"iat": datetime.now(timezone.utc),
100	"type": token_type,
101	})
102	encoded_jwt = jwt.encode(to_encode, secret_key, algorithm=ALGORITHM)	1✔
103	return encoded_jwt	1✔
104
105
106	def http_basic_or_jwt_token(form_data: HTTPBasicCredentials = Depends(httpbasic),	1✔
107	token: str = Depends(oauth2_scheme),
108	api_config=Depends(get_api_config)):
109	if token:	1✔
110	return get_user_from_token(token, api_config.get('jwt_secret_key', 'super-secret'))	1✔
111	elif form_data and verify_auth(api_config, form_data.username, form_data.password):	1✔
112	return form_data.username	1✔
113
114	raise HTTPException(	1✔
115	status_code=status.HTTP_401_UNAUTHORIZED,
116	detail="Unauthorized",
117	)
118
119
120	@router_login.post('/token/login', response_model=AccessAndRefreshToken)	1✔
121	def token_login(form_data: HTTPBasicCredentials = Depends(security),	1✔
122	api_config=Depends(get_api_config)):
123
124	if verify_auth(api_config, form_data.username, form_data.password):	1✔
125	token_data = {'identity': {'u': form_data.username}}	1✔
126	access_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'))	1✔
127	refresh_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'),	1✔
128	token_type="refresh")
129	return {	1✔
130	"access_token": access_token,
131	"refresh_token": refresh_token,
132	}
133	else:
134	raise HTTPException(	1✔
135	status_code=status.HTTP_401_UNAUTHORIZED,
136	detail="Incorrect username or password",
137	)
138
139
140	@router_login.post('/token/refresh', response_model=AccessToken)	1✔
141	def token_refresh(token: str = Depends(oauth2_scheme), api_config=Depends(get_api_config)):	1✔
142	# Refresh token
143	u = get_user_from_token(token, api_config.get(	1✔
144	'jwt_secret_key', 'super-secret'), 'refresh')
145	token_data = {'identity': {'u': u}}	1✔
146	access_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'),	1✔
147	token_type="access")
148	return {'access_token': access_token}	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous