kytos.core.auth.Auth._delete_user() - Code Metrics - Inspection of "feat: replaced `werkzeug/flask` with `uvicorn/star..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#375)

by Vinicius

created 2023-04-25 19:11 UTC

kytos.core.auth.Auth._delete_user() A

↳ Parent: kytos.core.auth

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	2
dl	0
loc	7
rs	10
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, 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 = self.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)

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

    @staticmethod
    def error_msg(error_list: list) -> str:
        """Return a more request friendly error message from ValidationError"""
        msg = ""
        for err in error_list:
            for value in err['loc']:
                msg += value + ", "
            msg = msg[:-2]
            msg += ": " + err["msg"] + "; "
        return msg[:-2]

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

kytos-ng / kytos

Pull Request — master (#375)

kytos.core.auth.Auth._delete_user() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like