10010184831

Committed 19 Jul 2024 03:04PM UTC coverage: 81.113% (-0.1%) from 81.219%

Build # 10010184831

Build Type

push

github

Committed by

web-flow

Commit Message

Get OIDC token using public authlib API (#221)

Avoids accessing internal/private token representation in authlib.

Run Details

313 of 412 branches covered (75.97%)

Branch coverage included in aggregate %.

8 of 8 new or added lines in 1 file covered. (100.0%)

2 existing lines in 1 file now uncovered.

984 of 1187 relevant lines covered (82.9%)

0.83 hits per line

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

81.13

/waiverdb/auth.py

# SPDX-License-Identifier: GPL-2.0+


import base64
import binascii
import gssapi
from flask import current_app, Request, Response, session
from werkzeug.exceptions import Unauthorized, Forbidden
from authlib.oauth2.base import OAuth2Error

from waiverdb.utils import auth_methods

OIDC_AUTH_HEADER_PREFIX = "Bearer "


# Inspired by https://github.com/mkomitee/flask-kerberos/blob/master/flask_kerberos.py
# Later cleaned and ported to python-gssapi
def process_gssapi_request(token):
    try:
        stage = "initialize server context"
        sc = gssapi.SecurityContext(usage="accept")

        stage = "step context"
        token = sc.step(token if token != "" else None)  # nosec
        token = token if token is not None else ""

        # The current architecture cannot support continuation here
        stage = "checking completion"
        if not sc.complete:
            current_app.logger.error(
                'Multiple GSSAPI round trips not supported')
            raise Forbidden("Attempted multiple GSSAPI round trips")

        current_app.logger.debug('Completed GSSAPI negotiation')

        stage = "getting remote user"
        user = str(sc.initiator_name)
        return user, token
    except gssapi.exceptions.GSSError as e:
        current_app.logger.exception(
            'Unable to authenticate: failed to %s: %s' %
            (stage, e.gen_message()))
        raise Forbidden("Authentication failed")


def get_user(request: Request) -> tuple[str, dict[str, str]]:
    methods = auth_methods(current_app)

    exceptions = []

    for method in methods:
        try:
            return get_user_by_method(request, method)
        except Unauthorized as e:
            exceptions.append(e)
            continue

    if exceptions:
        raise exceptions[0]
    raise Unauthorized("Authenticated user required. No methods specified.")


def get_oidc_userinfo(field: str) -> str:
    fields = session.get("oidc_auth_profile", {})
    if not fields:
        try:
            fields = current_app.oidc.accept_token.acquire_token()
        except OAuth2Error as e:
            raise Unauthorized(f"OIDC authentication failed: {e}")

    if field not in fields:
        current_app.logger.error(
            "User info field %r is unavailable; available are: %s", field, fields.keys()
        )
        raise Unauthorized("Failed to retrieve username")

    return fields[field]


def get_user_by_method(request: Request, auth_method: str) -> tuple[str, dict[str, str]]:
    user = None
    headers = dict()
    if auth_method == 'OIDC':
        user = get_oidc_userinfo(current_app.config['OIDC_USERNAME_FIELD'])
    elif auth_method == 'Kerberos':
        if 'Authorization' not in request.headers:
            response = Response('Unauthorized', 401, {'WWW-Authenticate': 'Negotiate'})
            raise Unauthorized(response=response)
        header = request.headers.get("Authorization")
        token = ''.join(header.strip().split()[1:])
        try:
            user, token = process_gssapi_request(base64.b64decode(token))
        except binascii.Error:
            raise Unauthorized("Invalid authentication token")
        # remove realm
        user = user.split("@")[0]
        headers = {'WWW-Authenticate': ' '.join(
            ['negotiate', base64.b64encode(token).decode()])}
    elif auth_method == 'SSL':
        # Nginx sets SSL_CLIENT_VERIFY and SSL_CLIENT_S_DN in request.environ
        # when doing SSL authentication.
        ssl_client_verify = request.environ.get('SSL_CLIENT_VERIFY')
        if ssl_client_verify != 'SUCCESS':
            raise Unauthorized('Cannot verify client: %s' % ssl_client_verify)
        if not request.environ.get('SSL_CLIENT_S_DN'):
            raise Unauthorized('Unable to get user information (DN) from the client certificate')
        user = request.environ.get('SSL_CLIENT_S_DN')
    elif auth_method == 'dummy':
        # Blindly accept any username. For testing purposes only of course!
        if not request.authorization:
            response = Response('Unauthorized', 401, {'WWW-Authenticate': 'Basic realm="dummy"'})
            raise Unauthorized(response=response)
        user = request.authorization.username
    else:
        raise Unauthorized(f"Unsupported authentication method {auth_method}")
    return user, headers

1	# SPDX-License-Identifier: GPL-2.0+
2
3
4	import base64	1✔
5	import binascii	1✔
6	import gssapi	1✔
7	from flask import current_app, Request, Response, session	1✔
8	from werkzeug.exceptions import Unauthorized, Forbidden	1✔
9	from authlib.oauth2.base import OAuth2Error	1✔
10
11	from waiverdb.utils import auth_methods	1✔
12
13	OIDC_AUTH_HEADER_PREFIX = "Bearer "	1✔
14
15
16	# Inspired by https://github.com/mkomitee/flask-kerberos/blob/master/flask_kerberos.py
17	# Later cleaned and ported to python-gssapi
18	def process_gssapi_request(token):	1✔
19	try:	1✔
20	stage = "initialize server context"	1✔
21	sc = gssapi.SecurityContext(usage="accept")	1✔
22
23	stage = "step context"	1✔
24	token = sc.step(token if token != "" else None) # nosec	1✔
25	token = token if token is not None else ""	1✔
26
27	# The current architecture cannot support continuation here
28	stage = "checking completion"	1✔
29	if not sc.complete:	1!
30	current_app.logger.error(	×
31	'Multiple GSSAPI round trips not supported')
32	raise Forbidden("Attempted multiple GSSAPI round trips")	×
33
34	current_app.logger.debug('Completed GSSAPI negotiation')	1✔
35
36	stage = "getting remote user"	1✔
37	user = str(sc.initiator_name)	1✔
38	return user, token	1✔
39	except gssapi.exceptions.GSSError as e:	×
40	current_app.logger.exception(	×
41	'Unable to authenticate: failed to %s: %s' %
42	(stage, e.gen_message()))
43	raise Forbidden("Authentication failed")	×
44
45
46	def get_user(request: Request) -> tuple[str, dict[str, str]]:	1✔
47	methods = auth_methods(current_app)	1✔
48
49	exceptions = []	1✔
50
51	for method in methods:	1✔
52	try:	1✔
53	return get_user_by_method(request, method)	1✔
54	except Unauthorized as e:	1✔
55	exceptions.append(e)	1✔
56	continue	1✔
57
58	if exceptions:	1✔
59	raise exceptions[0]	1✔
60	raise Unauthorized("Authenticated user required. No methods specified.")	1✔
61
62
63	def get_oidc_userinfo(field: str) -> str:	1✔
64	fields = session.get("oidc_auth_profile", {})	1✔
65	if not fields:	1✔
66	try:	1✔
67	fields = current_app.oidc.accept_token.acquire_token()	1✔
68	except OAuth2Error as e:	1✔
69	raise Unauthorized(f"OIDC authentication failed: {e}")	1✔
70
71	if field not in fields:	1!
UNCOV 72	current_app.logger.error(	×
73	"User info field %r is unavailable; available are: %s", field, fields.keys()
74	)
UNCOV 75	raise Unauthorized("Failed to retrieve username")	×
76
77	return fields[field]	1✔
78
79
80	def get_user_by_method(request: Request, auth_method: str) -> tuple[str, dict[str, str]]:	1✔
81	user = None	1✔
82	headers = dict()	1✔
83	if auth_method == 'OIDC':	1✔
84	user = get_oidc_userinfo(current_app.config['OIDC_USERNAME_FIELD'])	1✔
85	elif auth_method == 'Kerberos':	1✔
86	if 'Authorization' not in request.headers:	1✔
87	response = Response('Unauthorized', 401, {'WWW-Authenticate': 'Negotiate'})	1✔
88	raise Unauthorized(response=response)	1✔
89	header = request.headers.get("Authorization")	1✔
90	token = ''.join(header.strip().split()[1:])	1✔
91	try:	1✔
92	user, token = process_gssapi_request(base64.b64decode(token))	1✔
93	except binascii.Error:	1✔
94	raise Unauthorized("Invalid authentication token")	1✔
95	# remove realm
96	user = user.split("@")[0]	1✔
97	headers = {'WWW-Authenticate': ' '.join(	1✔
98	['negotiate', base64.b64encode(token).decode()])}
99	elif auth_method == 'SSL':	1!
100	# Nginx sets SSL_CLIENT_VERIFY and SSL_CLIENT_S_DN in request.environ
101	# when doing SSL authentication.
102	ssl_client_verify = request.environ.get('SSL_CLIENT_VERIFY')	1✔
103	if ssl_client_verify != 'SUCCESS':	1✔
104	raise Unauthorized('Cannot verify client: %s' % ssl_client_verify)	1✔
105	if not request.environ.get('SSL_CLIENT_S_DN'):	1✔
106	raise Unauthorized('Unable to get user information (DN) from the client certificate')	1✔
107	user = request.environ.get('SSL_CLIENT_S_DN')	1✔
108	elif auth_method == 'dummy':	×
109	# Blindly accept any username. For testing purposes only of course!
110	if not request.authorization:	×
111	response = Response('Unauthorized', 401, {'WWW-Authenticate': 'Basic realm="dummy"'})	×
112	raise Unauthorized(response=response)	×
113	user = request.authorization.username	×
114	else:
115	raise Unauthorized(f"Unsupported authentication method {auth_method}")	×
116	return user, headers	1✔

release-engineering / waiverdb / 10010184831

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