8665415066

Committed 12 Apr 2024 04:55PM UTC coverage: 52.282% (+1.1%) from 51.171%

Build # 8665415066

Build Type

push

github

Committed by

vokimon

Commit Message

Merge branch 'REF_kill_mithril_2'

Run Details

1055 of 1858 branches covered (56.78%)

Branch coverage included in aggregate %.

33 of 40 new or added lines in 4 files covered. (82.5%)

444 existing lines in 10 files now uncovered.

1797 of 3597 relevant lines covered (49.96%)

0.5 hits per line

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

68.75

/tomatic/auth.py

import json
import datetime
from functools import lru_cache
from fastapi import APIRouter, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer
from starlette.requests import Request
from starlette.responses import HTMLResponse, RedirectResponse
from starlette.config import Config
from authlib.integrations.starlette_client import OAuth, OAuthError
from jose import JWTError, jwt
from yamlns import namespace as ns
from consolemsg import error
from . import persons
import os
JWT_ALGORITHM='HS256'
CONF_URL = 'https://accounts.google.com/.well-known/openid-configuration'

def config(key='', default=Ellipsis):
    keyparts = key.split('.')
    envkey = key.replace('.','_').upper()
    if envkey in os.environ:
        return os.environ[envkey]
    try:
        import dbconfig as configuration
        configuration = ns(configuration.__dict__)
    except ImportError:
        configuration = ns(tomatic=ns())
    if not key:
        return configuration
    for part in keyparts[:-1]:
        configuration = configuration.get(part, ns())
    if default is Ellipsis and keyparts[-1] not in configuration:
        raise KeyError(key)
    return configuration.get(keyparts[-1], default)

@lru_cache
def oauth():
    oauth = OAuth()
    oauth.register(
        name='google',
        server_metadata_url=CONF_URL,
        client_kwargs={
            'scope': 'openid email profile'
        },
        **config('tomatic.oauth')
    )
    return oauth

router = APIRouter()

@router.get('/login')
async def login(request: Request):
    redirect_uri = str(request.url_for('auth'))
    if 'localhost' not in redirect_uri:
        redirect_uri = redirect_uri.replace('http:', 'https:')
    print("Auth redirect uri:", redirect_uri)
    return await oauth().google.authorize_redirect(request, redirect_uri)

def auth_result(token=None, error=None, code=200):
    return HTMLResponse(
        f""""<html><script>
        localStorage.setItem("token", "{token if token else ''}");
        localStorage.setItem("autherror", "{error if error else ''}");
        location.href="/";
        </script><h1>{error if error else ''}</h1></html>
        """, code)

@router.get('/auth')
async def auth(request: Request):
    try:
        token = await oauth().google.authorize_access_token(request)
    except OAuthError as error:
        return auth_result(error=f"Error d'autenticació: {error.error}", code=400)
    user = await oauth().google.parse_id_token(request, token)
    if not user:
        return auth_result(error='Error a la resposta de Google', code=400)

    username = persons.byEmail(user['email'])
    if not username:
        return auth_result(error=f"L'usuari {user['email']} no té accés a l'aplicació", code=400)

    user.update(username = username)
    token = create_access_token(user)
    return auth_result(token=token)

def expiration_timestamp(expiration_delta=None):
    expires_delta = expiration_delta or datetime.timedelta(
        **config('tomatic.jwt.expiration', dict(hours=10))
    )
    utcnow = datetime.datetime.now(datetime.timezone.utc)
    expiration = utcnow + expires_delta
    return int(expiration.timestamp())

@router.get('/logout')
async def logout(request: Request):
    request.session.pop('user', None)
    return RedirectResponse(url='/')

def create_access_token(data: dict, expiration_delta: datetime.timedelta = None):
    passthru_fields = (
        'username name email locale family_name given_name picture'
    ).split()
    payload = dict(
        (field, data[field])
        for field in passthru_fields
        if field in data
    )
    payload['exp'] = expiration_timestamp(expiration_delta)
    token = jwt.encode(
        payload,
        config('tomatic.jwt.secret_key'),
        algorithm=JWT_ALGORITHM,
    )
    return token

def auth_error(message):
    error(message)
    return HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=message or "Invalid authentication credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def validatedUser(token: str = Depends(oauth2_scheme)):
    environ_user = config('tomatic.auth.dummy', None)
    if environ_user:
        if environ_user.isalpha():
            return dict(
                username = environ_user,
                email = environ_user+'@somenergia.coop',
                exp = expiration_timestamp(),
            )
        return dict(
            username = 'alice',
            email = 'me@here.coop',
            exp = expiration_timestamp(),
        )
    try:
        payload = jwt.decode(
            token,
            config('tomatic.jwt.secret_key'),
            algorithms=JWT_ALGORITHM,
        )
    except JWTError as e:
        raise auth_error(f"Token decoding failed: {e}")

    # TODO: validate all the fields
    username: str = payload.get("username")
    if username is None:
        raise auth_error("Payload failed")

    return ns(payload)

def userInGroup(user, group):
    username = persons.byEmail(user['email'])
    groups = persons.persons().get('groups',{})
    return username in groups.get(group, [])

def adminUser(user: ns = Depends(validatedUser)):
    if not userInGroup(user, 'admin'):
        raise auth_error("Admin role required")
    return user

1	import json	1✔
2	import datetime	1✔
3	from functools import lru_cache	1✔
4	from fastapi import APIRouter, HTTPException, Depends, status	1✔
5	from fastapi.security import OAuth2PasswordBearer	1✔
6	from starlette.requests import Request	1✔
7	from starlette.responses import HTMLResponse, RedirectResponse	1✔
8	from starlette.config import Config	1✔
9	from authlib.integrations.starlette_client import OAuth, OAuthError	1✔
10	from jose import JWTError, jwt	1✔
11	from yamlns import namespace as ns	1✔
12	from consolemsg import error	1✔
13	from . import persons	1✔
14	import os	1✔
15	JWT_ALGORITHM='HS256'	1✔
16	CONF_URL = 'https://accounts.google.com/.well-known/openid-configuration'	1✔
17
18	def config(key='', default=Ellipsis):	1✔
19	keyparts = key.split('.')	1✔
20	envkey = key.replace('.','_').upper()	1✔
21	if envkey in os.environ:	1✔
22	return os.environ[envkey]	1✔
23	try:	1✔
24	import dbconfig as configuration	1✔
25	configuration = ns(configuration.__dict__)	×
26	except ImportError:	1✔
27	configuration = ns(tomatic=ns())	1✔
28	if not key:	1!
29	return configuration	×
30	for part in keyparts[:-1]:	1✔
31	configuration = configuration.get(part, ns())	1✔
32	if default is Ellipsis and keyparts[-1] not in configuration:	1!
33	raise KeyError(key)	×
34	return configuration.get(keyparts[-1], default)	1✔
35
36	@lru_cache	1✔
37	def oauth():	1✔
38	oauth = OAuth()	×
39	oauth.register(	×
40	name='google',
41	server_metadata_url=CONF_URL,
42	client_kwargs={
43	'scope': 'openid email profile'
44	},
45	**config('tomatic.oauth')
46	)
47	return oauth	×
48
49	router = APIRouter()	1✔
50
51	@router.get('/login')	1✔
52	async def login(request: Request):	1✔
53	redirect_uri = str(request.url_for('auth'))	×
54	if 'localhost' not in redirect_uri:	×
55	redirect_uri = redirect_uri.replace('http:', 'https:')	×
56	print("Auth redirect uri:", redirect_uri)	×
57	return await oauth().google.authorize_redirect(request, redirect_uri)	×
58
59	def auth_result(token=None, error=None, code=200):	1✔
60	return HTMLResponse(	×
61	f""""<html><script>
62	localStorage.setItem("token", "{token if token else ''}");
63	localStorage.setItem("autherror", "{error if error else ''}");
64	location.href="/";
65	</script><h1>{error if error else ''}</h1></html>
66	""", code)
67
68	@router.get('/auth')	1✔
69	async def auth(request: Request):	1✔
70	try:	×
71	token = await oauth().google.authorize_access_token(request)	×
72	except OAuthError as error:	×
73	return auth_result(error=f"Error d'autenticació: {error.error}", code=400)	×
74	user = await oauth().google.parse_id_token(request, token)	×
75	if not user:	×
76	return auth_result(error='Error a la resposta de Google', code=400)	×
77
78	username = persons.byEmail(user['email'])	×
79	if not username:	×
80	return auth_result(error=f"L'usuari {user['email']} no té accés a l'aplicació", code=400)	×
81
82	user.update(username = username)	×
83	token = create_access_token(user)	×
84	return auth_result(token=token)	×
85
86	def expiration_timestamp(expiration_delta=None):	1✔
87	expires_delta = expiration_delta or datetime.timedelta(	1✔
88	**config('tomatic.jwt.expiration', dict(hours=10))
89	)
90	utcnow = datetime.datetime.now(datetime.timezone.utc)	1✔
91	expiration = utcnow + expires_delta	1✔
92	return int(expiration.timestamp())	1✔
93
94	@router.get('/logout')	1✔
95	async def logout(request: Request):	1✔
96	request.session.pop('user', None)	×
97	return RedirectResponse(url='/')	×
98
99	def create_access_token(data: dict, expiration_delta: datetime.timedelta = None):	1✔
100	passthru_fields = (	1✔
101	'username name email locale family_name given_name picture'
102	).split()
103	payload = dict(	1✔
104	(field, data[field])
105	for field in passthru_fields
106	if field in data
107	)
108	payload['exp'] = expiration_timestamp(expiration_delta)	1✔
109	token = jwt.encode(	1✔
110	payload,
111	config('tomatic.jwt.secret_key'),
112	algorithm=JWT_ALGORITHM,
113	)
114	return token	1✔
115
116	def auth_error(message):	1✔
117	error(message)	1✔
118	return HTTPException(	1✔
119	status_code=status.HTTP_401_UNAUTHORIZED,
120	detail=message or "Invalid authentication credentials",
121	headers={"WWW-Authenticate": "Bearer"},
122	)
123
124
125	oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")	1✔
126
127	def validatedUser(token: str = Depends(oauth2_scheme)):	1✔
128	environ_user = config('tomatic.auth.dummy', None)	1✔
129	if environ_user:	1✔
130	if environ_user.isalpha():	1✔
131	return dict(	1✔
132	username = environ_user,
133	email = environ_user+'@somenergia.coop',
134	exp = expiration_timestamp(),
135	)
136	return dict(	1✔
137	username = 'alice',
138	email = 'me@here.coop',
139	exp = expiration_timestamp(),
140	)
141	try:	1✔
142	payload = jwt.decode(	1✔
143	token,
144	config('tomatic.jwt.secret_key'),
145	algorithms=JWT_ALGORITHM,
146	)
147	except JWTError as e:	1✔
148	raise auth_error(f"Token decoding failed: {e}")	1✔
149
150	# TODO: validate all the fields
151	username: str = payload.get("username")	1✔
152	if username is None:	1✔
153	raise auth_error("Payload failed")	1✔
154
155	return ns(payload)	1✔
156
157	def userInGroup(user, group):	1✔
158	username = persons.byEmail(user['email'])	1✔
159	groups = persons.persons().get('groups',{})	1✔
160	return username in groups.get(group, [])	1✔
161
162	def adminUser(user: ns = Depends(validatedUser)):	1✔
NEW 163	if not userInGroup(user, 'admin'):	×
UNCOV 164	raise auth_error("Admin role required")	×
UNCOV 165	return user	×

Som-Energia / somenergia-tomatic / 8665415066

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