kytos.core.auth.Auth._authenticate_user() - Code Metrics - Inspection of "blueprint for a message broker bridge NApp" - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#1267)

unknown

created 2024-03-15 19:57 UTC

kytos.core.auth.Auth._authenticate_user() B

↳ Parent: kytos.core.auth

Complexity

Conditions

Size

Total Lines	24
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	22
nop	2
dl	0
loc	24
rs	8.8853
c	0
b	0
f	0

"""Module with main classes related to Authentication."""

# pylint: disable=invalid-name
import base64
import binascii
import datetime
import getpass
import logging
import os
import uuid
from functools import wraps
from http import HTTPStatus

import jwt
import pymongo
from pydantic import ValidationError
from pymongo.collection import ReturnDocument
from pymongo.errors import AutoReconnect, DuplicateKeyError
from pymongo.results import InsertOneResult
from tenacity import retry_if_exception_type, stop_after_attempt, wait_random

from kytos.core.config import KytosConfig
from kytos.core.db import Mongo
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 content_type_json_or_415, error_msg,
                                 get_json_or_400)
from kytos.core.retry import before_sleep, for_all_methods, retries
from kytos.core.user import HashSubDoc, UserDoc, UserDocUpdate

__all__ = ['authenticated']

LOG = logging.getLogger(__name__)


def authenticated(func):
    """Handle tokens from requests."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        """Verify the requires of token."""
        try:
            request: Request = None
            for arg in args:
                if isinstance(arg, Request):
                    request = arg
                    break
            else:
                raise ValueError("Request arg not found in the decorated func")
            content = request.headers.get("Authorization")
            if content is None:
                raise ValueError("The attribute 'content' has an invalid "
                                 "value 'None'.")
            token = content.split("Bearer ")[1]
            jwt.decode(token, key=Auth.get_jwt_secret(),
                       algorithms=[Auth.encode_algorithm])
        except (
            ValueError,
            IndexError,
            jwt.exceptions.ExpiredSignatureError,
            jwt.exceptions.DecodeError,
        ) as exc:
            msg = f"Token not sent or expired: {exc}"
            return JSONResponse({"error": msg},
                                status_code=HTTPStatus.UNAUTHORIZED.value)
        return func(*args, **kwargs)

    return wrapper


@for_all_methods(
    retries,
    stop=stop_after_attempt(
        int(os.environ.get("MONGO_AUTO_RETRY_STOP_AFTER_ATTEMPT", 3))
    ),
    wait=wait_random(
        min=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MIN", 0.1)),
        max=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MAX", 1)),
    ),
    before_sleep=before_sleep,
    retry=retry_if_exception_type((AutoReconnect,)),
)
class UserController:
    """UserController"""

    # pylint: disable=unnecessary-lambda
    def __init__(self, get_mongo=lambda: Mongo()):

        self.mongo = get_mongo()
        self.db_client = self.mongo.client
        self.db = self.db_client[self.mongo.db_name]

    def bootstrap_indexes(self):
        """Bootstrap all users related indexes."""
        index_tuples = [
            ("users", [("username", pymongo.ASCENDING)], {"unique": True}),
        ]
        for collection, keys, kwargs in index_tuples:
            if self.mongo.bootstrap_index(collection, keys, **kwargs):
                LOG.info(
                    f"Created DB index {keys}, collection: {collection}"
                )

    def create_user(self, user_data: dict) -> InsertOneResult:
        """Create user to database"""
        try:
            utc_now = datetime.datetime.utcnow()
            result = self.db.users.insert_one(UserDoc(**{
                "_id": str(uuid.uuid4()),
                "username": user_data.get('username'),
                "hash": HashSubDoc(),
                "password": user_data.get('password'),
                "email": user_data.get('email'),
                "inserted_at": utc_now,
                "updated_at": utc_now,
            }).dict())
        except DuplicateKeyError as err:
            raise err
        except ValidationError as err:
            raise err
        return result

    def delete_user(self, username: str) -> dict:
        """Delete user from database"""
        utc_now = datetime.datetime.utcnow()
        result = self.db.users.find_one_and_update(
            {"username": username},
            {"$set": {
                "state": "inactive",
                "deleted_at": utc_now,
            }},
            return_document=ReturnDocument.AFTER,
        )
        return result

    def update_user(self, username: str, data: dict) -> dict:
        """Update user from database"""
        utc_now = datetime.datetime.utcnow()
        if "password" in data:
            data["hash"] = HashSubDoc()
        try:
            result = self.db.users.find_one_and_update(
                {"username": username},
                {
                    "$set": UserDocUpdate(**{
                        **data,
                        **{"updated_at": utc_now}
                    }).dict(exclude_none=True)
                },
                return_document=ReturnDocument.AFTER
            )
        except ValidationError as err:
            raise err
        except DuplicateKeyError as err:
            raise err
        return result

    def get_user(self, username: str) -> dict:
        """Return a user information from database"""
        data = self.db.users.aggregate([
            {"$match": {"username": username}},
            {"$project": UserDoc.projection()},
            {"$limit": 1}
        ])
        try:
            user, *_ = list(value for value in data)
            return user
        except ValueError:
            return {}

    def get_user_nopw(self, username: str) -> dict:
        """Return a user information from database without password"""
        data = self.db.users.aggregate([
            {"$match": {"username": username}},
            {"$project": UserDoc.projection_nopw()},
            {"$limit": 1}
        ])
        try:
            user, *_ = list(value for value in data)
            return user
        except ValueError:
            return {}

    def get_users(self) -> dict:
        """Return all the users"""
        data = self.db.users.aggregate([
            {"$project": UserDoc.projection_nopw()}
        ])
        return {'users': list(value for value in data)}


class Auth:
    """Module used to provide Kytos authentication routes."""

    encode_algorithm = "HS256"

    def __init__(self, controller):
        """Init method of Auth class takes the parameters below.

        Args:
            controller(kytos.core.controller): A Controller instance.

        """
        self.user_controller = self.get_user_controller()
        self.user_controller.bootstrap_indexes()
        self.controller = controller
        self.token_expiration_minutes = self.get_token_expiration()
        if self.controller.options.create_superuser is True:
            self._create_superuser()

    @staticmethod
    def get_user_controller():
        """Get UserController"""
        return UserController()

    @staticmethod
    def get_token_expiration():
        """Return token expiration time in minutes defined in kytos conf."""
        options = KytosConfig().options['daemon']
        return options.token_expiration_minutes

    @classmethod
    def get_jwt_secret(cls):
        """Return JWT secret defined in kytos conf."""
        options = KytosConfig().options['daemon']
        return options.jwt_secret

    @classmethod
    def _generate_token(cls, username, time_exp):
        """Generate a jwt token."""
        return jwt.encode(
            {
                'username': username,
                'iss': "Kytos NApps Server",
                'exp': time_exp,
            },
            Auth.get_jwt_secret(),
            algorithm=cls.encode_algorithm,
        )

    def _create_superuser(self):
        """Create a superuser using MongoDB."""
        def get_username():
            return input("Username: ")

        def get_email():
            return input("Email: ")

        username = get_username()
        email = get_email()
        while True:
            password = getpass.getpass()
            re_password = getpass.getpass('Retype password: ')
            if password == re_password:
                break
        user = {
            "username": username,
            "email": email,
            "password": password,
        }
        try:
            self.user_controller.create_user(user)
        except ValidationError as err:
            msg = error_msg(err.errors())
            return f"Error: {msg}"
        except DuplicateKeyError:
            return f"Error: {username} already exist."

        return "User successfully created"

    def register_core_auth_services(self):
        """
        Register /kytos/core/ services over authentication.

        It registers create, authenticate, list all, list specific, delete and
        update users.
        """
        self.controller.api_server.register_core_endpoint(
            "auth/login/", self._authenticate_user
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/", self._list_users,
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/{username}", self._list_user
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/", self._create_user, methods=["POST"]
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/{username}", self._delete_user, methods=["DELETE"]
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/{username}", self._update_user, methods=["PATCH"]
        )

    def _authenticate_user(self, request: Request) -> JSONResponse:
        """Authenticate a user using MongoDB."""
        if "Authorization" not in request.headers:
            raise HTTPException(400, detail="Authorization header missing")
        try:
            auth = request.headers["Authorization"]
            _scheme, credentials = auth.split()
            decoded = base64.b64decode(credentials).decode("ascii")
            username, _, password = decoded.partition(":")
            password = password.encode()
        except (ValueError, TypeError, binascii.Error) as err:
            msg = "Credentials were not correctly set."
            raise HTTPException(400, detail=msg) from err
        user = self._find_user(username)
        if user["state"] != 'active':
            raise HTTPException(401, detail='This user is not active')
        password_hashed = UserDoc.hashing(password, user["hash"])
        if user["password"] != password_hashed:
            raise HTTPException(401, detail="Incorrect password")
        time_exp = datetime.datetime.utcnow() + datetime.timedelta(
            minutes=self.token_expiration_minutes
        )
        token = self._generate_token(username, time_exp)
        return JSONResponse({"token": token})

    def _find_user(self, username):
        """Find a specific user using MongoDB."""
        user = self.user_controller.get_user(username)
        if not user:
            raise HTTPException(404, detail=f"User {username} not found")
        return user

    @authenticated
    def _list_user(self, request: Request) -> JSONResponse:
        """List a specific user using MongoDB."""
        username = request.path_params["username"]
        user = self.user_controller.get_user_nopw(username)
        if not user:
            raise HTTPException(404, detail=f"User {username} not found")
        return JSONResponse(user)

    @authenticated
    def _list_users(self, _request: Request) -> JSONResponse:
        """List all users using MongoDB."""
        users_list = self.user_controller.get_users()
        return JSONResponse(users_list)

    def _get_request_body(self, request: Request) -> dict:
        """Get JSON from request"""
        content_type_json_or_415(request)
        body = get_json_or_400(request, self.controller.loop)
        if not isinstance(body, dict):
            raise HTTPException(400, "Invalid payload type: {body}")
        return body

    @authenticated
    def _create_user(self, request: Request) -> JSONResponse:
        """Save a user using MongoDB."""
        try:
            user_data = self._get_request_body(request)
            self.user_controller.create_user(user_data)
        except ValidationError as err:
            msg = error_msg(err.errors())
            raise HTTPException(400, msg) from err
        except DuplicateKeyError as err:
            msg = user_data.get("username") + " already exists."
            raise HTTPException(409, detail=msg) from err
        return JSONResponse("User successfully created", status_code=201)

    @authenticated
    def _delete_user(self, request: Request) -> JSONResponse:
        """Delete a user using MongoDB."""
        username = request.path_params["username"]
        if not self.user_controller.delete_user(username):
            raise HTTPException(404, detail=f"User {username} not found")
        return JSONResponse(f"User {username} deleted succesfully")

    @authenticated
    def _update_user(self, request: Request) -> JSONResponse:
        """Update user data using MongoDB."""
        body = self._get_request_body(request)
        username = request.path_params["username"]
        allowed = ["username", "email", "password"]
        data = {}
        for key, value in body.items():
            if key in allowed:
                data[key] = value
        try:
            updated = self.user_controller.update_user(username, data)
        except ValidationError as err:
            msg = error_msg(err.errors())
            raise HTTPException(400, detail=msg) from err
        except DuplicateKeyError as err:
            msg = username + " already exists."
            raise HTTPException(409, detail=msg) from err
        if not updated:
            raise HTTPException(404, detail=f"User {username} not found")
        return JSONResponse("User successfully updated")


1			"""Module with main classes related to Authentication."""
2
3			# pylint: disable=invalid-name
4			import base64
5			import binascii
6			import datetime
7			import getpass
8			import logging
9			import os
10			import uuid
11			from functools import wraps
12			from http import HTTPStatus
13
14			import jwt
15			import pymongo
16			from pydantic import ValidationError
17			from pymongo.collection import ReturnDocument
18			from pymongo.errors import AutoReconnect, DuplicateKeyError
19			from pymongo.results import InsertOneResult
20			from tenacity import retry_if_exception_type, stop_after_attempt, wait_random
21
22			from kytos.core.config import KytosConfig
23			from kytos.core.db import Mongo
24			from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
25			content_type_json_or_415, error_msg,
26			get_json_or_400)
27			from kytos.core.retry import before_sleep, for_all_methods, retries
28			from kytos.core.user import HashSubDoc, UserDoc, UserDocUpdate
29
30			__all__ = ['authenticated']
31
32			LOG = logging.getLogger(__name__)
33
34
35			def authenticated(func):
36			"""Handle tokens from requests."""
37			@wraps(func)
38			def wrapper(args, *kwargs):
39			"""Verify the requires of token."""
40			try:
41			request: Request = None
42			for arg in args:
43			if isinstance(arg, Request):
44			request = arg
45			break
46			else:
47			raise ValueError("Request arg not found in the decorated func")
48			content = request.headers.get("Authorization")
49			if content is None:
50			raise ValueError("The attribute 'content' has an invalid "
51			"value 'None'.")
52			token = content.split("Bearer ")[1]
53			jwt.decode(token, key=Auth.get_jwt_secret(),
54			algorithms=[Auth.encode_algorithm])
55			except (
56			ValueError,
57			IndexError,
58			jwt.exceptions.ExpiredSignatureError,
59			jwt.exceptions.DecodeError,
60			) as exc:
61			msg = f"Token not sent or expired: {exc}"
62			return JSONResponse({"error": msg},
63			status_code=HTTPStatus.UNAUTHORIZED.value)
64			return func(args, *kwargs)
65
66			return wrapper
67
68
69			@for_all_methods(
70			retries,
71			stop=stop_after_attempt(
72			int(os.environ.get("MONGO_AUTO_RETRY_STOP_AFTER_ATTEMPT", 3))
73			),
74			wait=wait_random(
75			min=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MIN", 0.1)),
76			max=int(os.environ.get("MONGO_AUTO_RETRY_WAIT_RANDOM_MAX", 1)),
77			),
78			before_sleep=before_sleep,
79			retry=retry_if_exception_type((AutoReconnect,)),
80			)
81			class UserController:
82			"""UserController"""
83
84			# pylint: disable=unnecessary-lambda
85			def __init__(self, get_mongo=lambda: Mongo()):
			0 ignored issues – show Comprehensibility Best Practice introduced 2024-03-15 20:06 UTC by Report Bug Copy Issue Report The variable `Mongo` does not seem to be defined. Loading history...
86			self.mongo = get_mongo()
87			self.db_client = self.mongo.client
88			self.db = self.db_client[self.mongo.db_name]
89
90			def bootstrap_indexes(self):
91			"""Bootstrap all users related indexes."""
92			index_tuples = [
93			("users", [("username", pymongo.ASCENDING)], {"unique": True}),
94			]
95			for collection, keys, kwargs in index_tuples:
96			if self.mongo.bootstrap_index(collection, keys, **kwargs):
97			LOG.info(
98			f"Created DB index {keys}, collection: {collection}"
99			)
100
101			def create_user(self, user_data: dict) -> InsertOneResult:
102			"""Create user to database"""
103			try:
104			utc_now = datetime.datetime.utcnow()
105			result = self.db.users.insert_one(UserDoc(**{
106			"_id": str(uuid.uuid4()),
107			"username": user_data.get('username'),
108			"hash": HashSubDoc(),
109			"password": user_data.get('password'),
110			"email": user_data.get('email'),
111			"inserted_at": utc_now,
112			"updated_at": utc_now,
113			}).dict())
114			except DuplicateKeyError as err:
115			raise err
116			except ValidationError as err:
117			raise err
118			return result
119
120			def delete_user(self, username: str) -> dict:
121			"""Delete user from database"""
122			utc_now = datetime.datetime.utcnow()
123			result = self.db.users.find_one_and_update(
124			{"username": username},
125			{"$set": {
126			"state": "inactive",
127			"deleted_at": utc_now,
128			}},
129			return_document=ReturnDocument.AFTER,
130			)
131			return result
132
133			def update_user(self, username: str, data: dict) -> dict:
134			"""Update user from database"""
135			utc_now = datetime.datetime.utcnow()
136			if "password" in data:
137			data["hash"] = HashSubDoc()
138			try:
139			result = self.db.users.find_one_and_update(
140			{"username": username},
141			{
142			"$set": UserDocUpdate(**{
143			**data,
144			**{"updated_at": utc_now}
145			}).dict(exclude_none=True)
146			},
147			return_document=ReturnDocument.AFTER
148			)
149			except ValidationError as err:
150			raise err
151			except DuplicateKeyError as err:
152			raise err
153			return result
154
155			def get_user(self, username: str) -> dict:
156			"""Return a user information from database"""
157			data = self.db.users.aggregate([
158			{"$match": {"username": username}},
159			{"$project": UserDoc.projection()},
160			{"$limit": 1}
161			])
162			try:
163			user, *_ = list(value for value in data)
164			return user
165			except ValueError:
166			return {}
167
168			def get_user_nopw(self, username: str) -> dict:
169			"""Return a user information from database without password"""
170			data = self.db.users.aggregate([
171			{"$match": {"username": username}},
172			{"$project": UserDoc.projection_nopw()},
173			{"$limit": 1}
174			])
175			try:
176			user, *_ = list(value for value in data)
177			return user
178			except ValueError:
179			return {}
180
181			def get_users(self) -> dict:
182			"""Return all the users"""
183			data = self.db.users.aggregate([
184			{"$project": UserDoc.projection_nopw()}
185			])
186			return {'users': list(value for value in data)}
187
188
189			class Auth:
190			"""Module used to provide Kytos authentication routes."""
191
192			encode_algorithm = "HS256"
193
194			def __init__(self, controller):
195			"""Init method of Auth class takes the parameters below.
196
197			Args:
198			controller(kytos.core.controller): A Controller instance.
199
200			"""
201			self.user_controller = self.get_user_controller()
202			self.user_controller.bootstrap_indexes()
203			self.controller = controller
204			self.token_expiration_minutes = self.get_token_expiration()
205			if self.controller.options.create_superuser is True:
206			self._create_superuser()
207
208			@staticmethod
209			def get_user_controller():
210			"""Get UserController"""
211			return UserController()
212
213			@staticmethod
214			def get_token_expiration():
215			"""Return token expiration time in minutes defined in kytos conf."""
216			options = KytosConfig().options['daemon']
217			return options.token_expiration_minutes
218
219			@classmethod
220			def get_jwt_secret(cls):
221			"""Return JWT secret defined in kytos conf."""
222			options = KytosConfig().options['daemon']
223			return options.jwt_secret
224
225			@classmethod
226			def _generate_token(cls, username, time_exp):
227			"""Generate a jwt token."""
228			return jwt.encode(
229			{
230			'username': username,
231			'iss': "Kytos NApps Server",
232			'exp': time_exp,
233			},
234			Auth.get_jwt_secret(),
235			algorithm=cls.encode_algorithm,
236			)
237
238			def _create_superuser(self):
239			"""Create a superuser using MongoDB."""
240			def get_username():
241			return input("Username: ")
242
243			def get_email():
244			return input("Email: ")
245
246			username = get_username()
247			email = get_email()
248			while True:
249			password = getpass.getpass()
250			re_password = getpass.getpass('Retype password: ')
251			if password == re_password:
252			break
253			user = {
254			"username": username,
255			"email": email,
256			"password": password,
257			}
258			try:
259			self.user_controller.create_user(user)
260			except ValidationError as err:
261			msg = error_msg(err.errors())
262			return f"Error: {msg}"
263			except DuplicateKeyError:
264			return f"Error: {username} already exist."
265
266			return "User successfully created"
267
268			def register_core_auth_services(self):
269			"""
270			Register /kytos/core/ services over authentication.
271
272			It registers create, authenticate, list all, list specific, delete and
273			update users.
274			"""
275			self.controller.api_server.register_core_endpoint(
276			"auth/login/", self._authenticate_user
277			)
278			self.controller.api_server.register_core_endpoint(
279			"auth/users/", self._list_users,
280			)
281			self.controller.api_server.register_core_endpoint(
282			"auth/users/{username}", self._list_user
283			)
284			self.controller.api_server.register_core_endpoint(
285			"auth/users/", self._create_user, methods=["POST"]
286			)
287			self.controller.api_server.register_core_endpoint(
288			"auth/users/{username}", self._delete_user, methods=["DELETE"]
289			)
290			self.controller.api_server.register_core_endpoint(
291			"auth/users/{username}", self._update_user, methods=["PATCH"]
292			)
293
294			def _authenticate_user(self, request: Request) -> JSONResponse:
295			"""Authenticate a user using MongoDB."""
296			if "Authorization" not in request.headers:
297			raise HTTPException(400, detail="Authorization header missing")
298			try:
299			auth = request.headers["Authorization"]
300			_scheme, credentials = auth.split()
301			decoded = base64.b64decode(credentials).decode("ascii")
302			username, _, password = decoded.partition(":")
303			password = password.encode()
304			except (ValueError, TypeError, binascii.Error) as err:
305			msg = "Credentials were not correctly set."
306			raise HTTPException(400, detail=msg) from err
307			user = self._find_user(username)
308			if user["state"] != 'active':
309			raise HTTPException(401, detail='This user is not active')
310			password_hashed = UserDoc.hashing(password, user["hash"])
311			if user["password"] != password_hashed:
312			raise HTTPException(401, detail="Incorrect password")
313			time_exp = datetime.datetime.utcnow() + datetime.timedelta(
314			minutes=self.token_expiration_minutes
315			)
316			token = self._generate_token(username, time_exp)
317			return JSONResponse({"token": token})
318
319			def _find_user(self, username):
320			"""Find a specific user using MongoDB."""
321			user = self.user_controller.get_user(username)
322			if not user:
323			raise HTTPException(404, detail=f"User {username} not found")
324			return user
325
326			@authenticated
327			def _list_user(self, request: Request) -> JSONResponse:
328			"""List a specific user using MongoDB."""
329			username = request.path_params["username"]
330			user = self.user_controller.get_user_nopw(username)
331			if not user:
332			raise HTTPException(404, detail=f"User {username} not found")
333			return JSONResponse(user)
334
335			@authenticated
336			def _list_users(self, _request: Request) -> JSONResponse:
337			"""List all users using MongoDB."""
338			users_list = self.user_controller.get_users()
339			return JSONResponse(users_list)
340
341			def _get_request_body(self, request: Request) -> dict:
342			"""Get JSON from request"""
343			content_type_json_or_415(request)
344			body = get_json_or_400(request, self.controller.loop)
345			if not isinstance(body, dict):
346			raise HTTPException(400, "Invalid payload type: {body}")
347			return body
348
349			@authenticated
350			def _create_user(self, request: Request) -> JSONResponse:
351			"""Save a user using MongoDB."""
352			try:
353			user_data = self._get_request_body(request)
354			self.user_controller.create_user(user_data)
355			except ValidationError as err:
356			msg = error_msg(err.errors())
357			raise HTTPException(400, msg) from err
358			except DuplicateKeyError as err:
359			msg = user_data.get("username") + " already exists."
360			raise HTTPException(409, detail=msg) from err
361			return JSONResponse("User successfully created", status_code=201)
362
363			@authenticated
364			def _delete_user(self, request: Request) -> JSONResponse:
365			"""Delete a user using MongoDB."""
366			username = request.path_params["username"]
367			if not self.user_controller.delete_user(username):
368			raise HTTPException(404, detail=f"User {username} not found")
369			return JSONResponse(f"User {username} deleted succesfully")
370
371			@authenticated
372			def _update_user(self, request: Request) -> JSONResponse:
373			"""Update user data using MongoDB."""
374			body = self._get_request_body(request)
375			username = request.path_params["username"]
376			allowed = ["username", "email", "password"]
377			data = {}
378			for key, value in body.items():
379			if key in allowed:
380			data[key] = value
381			try:
382			updated = self.user_controller.update_user(username, data)
383			except ValidationError as err:
384			msg = error_msg(err.errors())
385			raise HTTPException(400, detail=msg) from err
386			except DuplicateKeyError as err:
387			msg = username + " already exists."
388			raise HTTPException(409, detail=msg) from err
389			if not updated:
390			raise HTTPException(404, detail=f"User {username} not found")
391			return JSONResponse("User successfully updated")
392

kytos / kytos

Pull Request — master (#1267)

kytos.core.auth.Auth._authenticate_user() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like