kytos.core.auth.Auth._delete_user() - Code Metrics - Inspection of "[Feature][Fix] Upgrade dependencies" - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#284)

by Vinicius

created 2022-09-08 21:06 UTC

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

↳ Parent: kytos.core.auth

Complexity

Conditions

Size

Total Lines	35
Code Lines	27

Duplication

Lines	35
Ratio	100 %

Code Coverage

Tests	15
CRAP Score	5.0061

Importance

Changes

Metric	Value
cc	5
eloc	27
nop	2
dl	35
loc	35
rs	8.7653
c	0
b	0
f	0
ccs	15
cts	16
cp	0.9375
crap	5.0061

"""Module with main classes related to Authentication."""
import datetime
import getpass
import hashlib
import logging
import time
from functools import wraps
from http import HTTPStatus

import jwt
from flask import jsonify, request

from kytos.core.config import KytosConfig
from kytos.core.events import KytosEvent

__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:
            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 jsonify({"error": msg}), HTTPStatus.UNAUTHORIZED.value
        return func(*args, **kwargs)

    return wrapper


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.controller = controller
        self.namespace = "kytos.core.auth.users"
        self.token_expiration_minutes = self.get_token_expiration()
        if self.controller.options.create_superuser is True:
            self._create_superuser()

    @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 Storehouse."""
        def _create_superuser_callback(_event, box, error):
            if error:
                LOG.error('Superuser was not created. Error: %s', error)
            if box:
                LOG.info("Superuser successfully created")

        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
            print('Passwords do not match. Try again')

        user = {
            "username": username,
            "email": email,
            "password": hashlib.sha512(password.encode()).hexdigest(),
        }
        content = {
            "namespace": self.namespace,
            "box_id": user["username"],
            "data": user,
            "callback": _create_superuser_callback,
        }
        event = KytosEvent(name="kytos.storehouse.create", content=content)
        self.controller.buffers.app.put(event)

    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/<uid>", 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/<uid>", self._delete_user, methods=["DELETE"]
        )
        self.controller.api_server.register_core_endpoint(
            "auth/users/<uid>", self._update_user, methods=["PATCH"]
        )

    def _authenticate_user(self):
        """Authenticate a user using Storehouse."""
        username = request.authorization["username"]
        password = request.authorization["password"].encode()
        try:
            user = self._find_user(username)[0].get("data")
            if user.get("password") != hashlib.sha512(password).hexdigest():
                raise KeyError
            time_exp = datetime.datetime.utcnow() + datetime.timedelta(
                minutes=self.token_expiration_minutes
            )
            token = self._generate_token(username, time_exp)
            return {"token": token}, HTTPStatus.OK.value
        except (AttributeError, KeyError) as exc:
            result = f"Incorrect username or password: {exc}"
            return result, HTTPStatus.UNAUTHORIZED.value

    def _find_user(self, uid):
        """Find a specific user using Storehouse."""
        response = {}

        def _find_user_callback(_event, box, error):
            nonlocal response

            if not box:
                response = {
                    "answer": f'User with uid {uid} not found',
                    "code": HTTPStatus.NOT_FOUND.value
                }
            elif error:
                response = {
                    "answer": "User data cannot be shown",
                    "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
                }
            else:
                response = {
                    "answer": {"data": box.data},
                    "code": HTTPStatus.OK.value,
                }

        content = {
            "box_id": uid,
            "namespace": self.namespace,
            "callback": _find_user_callback,
        }
        event = KytosEvent(name="kytos.storehouse.retrieve", content=content)
        self.controller.buffers.app.put(event)
        while True:
            time.sleep(0.1)
            if response:
                break
        return response["answer"], response["code"]

    @authenticated
    def _list_user(self, uid):
        """List a specific user using Storehouse."""
        answer, code = self._find_user(uid)
        if code == HTTPStatus.OK.value:
            del answer['data']['password']
        return answer, code

    @authenticated

    def _list_users(self):
        """List all users using Storehouse."""
        response = {}

        def _list_users_callback(_event, boxes, error):
            nonlocal response

            if error:
                response = {
                    "answer": "Users cannot be listed",
                    "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
                }
            else:
                response = {
                    "answer": {"users": boxes},
                    "code": HTTPStatus.OK.value,
                }

        content = {
            "namespace": self.namespace,
            "callback": _list_users_callback,
        }
        event = KytosEvent(name="kytos.storehouse.list", content=content)
        self.controller.buffers.app.put(event)
        while True:
            time.sleep(0.1)
            if response:
                break
        return response["answer"], response["code"]

    @authenticated
    def _create_user(self):
        """Save a user using Storehouse."""
        response = {}

        def _create_user_callback(_event, box, error):
            nonlocal response

            if not box:
                response = {
                    "answer": 'User already exists',
                    "code": HTTPStatus.CONFLICT.value,
                }
            elif error:
                response = {
                    "answer": "User has not been created",
                    "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
                }
            else:
                response = {
                    "answer": "User successfully created",
                    "code": HTTPStatus.OK.value,
                }

        req = request.json
        password = req["password"].encode()
        data = {
            "username": req["username"],
            "email": req["email"],
            "password": hashlib.sha512(password).hexdigest(),
        }
        content = {
            "namespace": self.namespace,
            "box_id": data["username"],
            "data": data,
            "callback": _create_user_callback,
        }
        event = KytosEvent(name="kytos.storehouse.create", content=content)
        self.controller.buffers.app.put(event)
        while True:
            time.sleep(0.1)
            if response:
                break
        return response["answer"], response["code"]

    @authenticated

    def _delete_user(self, uid):
        """Delete a user using Storehouse."""
        response = {}

        def _delete_user_callback(_event, box, error):
            nonlocal response

            if not box:
                response = {
                    "answer": f'User with uid {uid} not found',
                    "code": HTTPStatus.NOT_FOUND.value
                }
            elif error:
                response = {
                    "answer": "User has not been deleted",
                    "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
                }
            else:
                response = {
                    "answer": "User successfully deleted",
                    "code": HTTPStatus.OK.value,
                }

        content = {
            "box_id": uid,
            "namespace": self.namespace,
            "callback": _delete_user_callback,
        }
        event = KytosEvent(name="kytos.storehouse.delete", content=content)
        self.controller.buffers.app.put(event)
        while True:
            time.sleep(0.1)
            if response:
                break
        return response["answer"], response["code"]

    @authenticated
    def _update_user(self, uid):
        """Update user data using Storehouse."""
        response = {}

        def _update_user_callback(_event, box, error):
            nonlocal response

            if not box:
                response = {
                    "answer": f'User with uid {uid} not found',
                    "code": HTTPStatus.NOT_FOUND.value
                }
            elif error:
                response = {
                    "answer": "User has not been updated",
                    "code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
                }
            else:
                response = {
                    "answer": "User successfully updated",
                    "code": HTTPStatus.OK.value,
                }

        req = request.json
        allowed = ["username", "email", "password"]

        data = {}
        for key, value in req.items():
            if key in allowed:
                data[key] = value

        content = {
            "namespace": self.namespace,
            "box_id": uid,
            "data": data,
            "callback": _update_user_callback,
        }
        event = KytosEvent(name="kytos.storehouse.update", content=content)
        self.controller.buffers.app.put(event)
        while True:
            time.sleep(0.1)
            if response:
                break
        return response["answer"], response["code"]


1			"""Module with main classes related to Authentication."""
2	1		import datetime
3	1		import getpass
4	1		import hashlib
5	1		import logging
6	1		import time
7	1		from functools import wraps
8	1		from http import HTTPStatus
9
10	1		import jwt
11	1		from flask import jsonify, request
12
13	1		from kytos.core.config import KytosConfig
14	1		from kytos.core.events import KytosEvent
15
16	1		__all__ = ['authenticated']
17
18	1		LOG = logging.getLogger(__name__)
19
20
21	1		def authenticated(func):
22			"""Handle tokens from requests."""
23	1		@wraps(func)
24	1		def wrapper(args, *kwargs):
25			"""Verify the requires of token."""
26	1		try:
27	1		content = request.headers.get("Authorization")
28	1		if content is None:
29			raise ValueError("The attribute 'content' has an invalid "
30			"value 'None'.")
31	1		token = content.split("Bearer ")[1]
32	1		jwt.decode(token, key=Auth.get_jwt_secret(),
33			algorithms=[Auth.encode_algorithm])
34	1		except (
35			ValueError,
36			IndexError,
37			jwt.exceptions.ExpiredSignatureError,
38			jwt.exceptions.DecodeError,
39			) as exc:
40	1		msg = f"Token not sent or expired: {exc}"
41	1		return jsonify({"error": msg}), HTTPStatus.UNAUTHORIZED.value
42	1		return func(args, *kwargs)
43
44	1		return wrapper
45
46
47	1		class Auth:
48			"""Module used to provide Kytos authentication routes."""
49
50	1		encode_algorithm = "HS256"
51
52	1		def __init__(self, controller):
53			"""Init method of Auth class takes the parameters below.
54
55			Args:
56			controller(kytos.core.controller): A Controller instance.
57
58			"""
59	1		self.controller = controller
60	1		self.namespace = "kytos.core.auth.users"
61	1		self.token_expiration_minutes = self.get_token_expiration()
62	1		if self.controller.options.create_superuser is True:
63			self._create_superuser()
64
65	1		@staticmethod
66	1		def get_token_expiration():
67			"""Return token expiration time in minutes defined in kytos conf."""
68	1		options = KytosConfig().options['daemon']
69	1		return options.token_expiration_minutes
70
71	1		@classmethod
72	1		def get_jwt_secret(cls):
73			"""Return JWT secret defined in kytos conf."""
74			options = KytosConfig().options['daemon']
75			return options.jwt_secret
76
77	1		@classmethod
78	1		def _generate_token(cls, username, time_exp):
79			"""Generate a jwt token."""
80	1		return jwt.encode(
81			{
82			'username': username,
83			'iss': "Kytos NApps Server",
84			'exp': time_exp,
85			},
86			Auth.get_jwt_secret(),
87			algorithm=cls.encode_algorithm,
88			)
89
90	1		def _create_superuser(self):
91			"""Create a superuser using Storehouse."""
92			def _create_superuser_callback(_event, box, error):
93			if error:
94			LOG.error('Superuser was not created. Error: %s', error)
95			if box:
96			LOG.info("Superuser successfully created")
97
98			def get_username():
99			return input("Username: ")
100
101			def get_email():
102			return input("Email: ")
103
104			username = get_username()
105			email = get_email()
106
107			while True:
108			password = getpass.getpass()
109			re_password = getpass.getpass('Retype password: ')
110			if password == re_password:
111			break
112			print('Passwords do not match. Try again')
113
114			user = {
115			"username": username,
116			"email": email,
117			"password": hashlib.sha512(password.encode()).hexdigest(),
118			}
119			content = {
120			"namespace": self.namespace,
121			"box_id": user["username"],
122			"data": user,
123			"callback": _create_superuser_callback,
124			}
125			event = KytosEvent(name="kytos.storehouse.create", content=content)
126			self.controller.buffers.app.put(event)
127
128	1		def register_core_auth_services(self):
129			"""
130			Register /kytos/core/ services over authentication.
131
132			It registers create, authenticate, list all, list specific, delete and
133			update users.
134			"""
135	1		self.controller.api_server.register_core_endpoint(
136			"auth/login/", self._authenticate_user
137			)
138	1		self.controller.api_server.register_core_endpoint(
139			"auth/users/", self._list_users
140			)
141	1		self.controller.api_server.register_core_endpoint(
142			"auth/users/<uid>", self._list_user
143			)
144	1		self.controller.api_server.register_core_endpoint(
145			"auth/users/", self._create_user, methods=["POST"]
146			)
147	1		self.controller.api_server.register_core_endpoint(
148			"auth/users/<uid>", self._delete_user, methods=["DELETE"]
149			)
150	1		self.controller.api_server.register_core_endpoint(
151			"auth/users/<uid>", self._update_user, methods=["PATCH"]
152			)
153
154	1		def _authenticate_user(self):
155			"""Authenticate a user using Storehouse."""
156	1		username = request.authorization["username"]
157	1		password = request.authorization["password"].encode()
158	1		try:
159	1		user = self._find_user(username)[0].get("data")
160	1		if user.get("password") != hashlib.sha512(password).hexdigest():
161	1		raise KeyError
162	1		time_exp = datetime.datetime.utcnow() + datetime.timedelta(
163			minutes=self.token_expiration_minutes
164			)
165	1		token = self._generate_token(username, time_exp)
166	1		return {"token": token}, HTTPStatus.OK.value
167	1		except (AttributeError, KeyError) as exc:
168	1		result = f"Incorrect username or password: {exc}"
169	1		return result, HTTPStatus.UNAUTHORIZED.value
170
171	1		def _find_user(self, uid):
172			"""Find a specific user using Storehouse."""
173	1		response = {}
174
175	1		def _find_user_callback(_event, box, error):
176			nonlocal response
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-11-30 19:20 UTC by Report Bug Copy Issue Report The variable `response` does not seem to be defined. Loading history...
177	1		if not box:
178	1		response = {
179			"answer": f'User with uid {uid} not found',
180			"code": HTTPStatus.NOT_FOUND.value
181			}
182	1		elif error:
183			response = {
184			"answer": "User data cannot be shown",
185			"code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
186			}
187			else:
188	1		response = {
189			"answer": {"data": box.data},
190			"code": HTTPStatus.OK.value,
191			}
192
193	1		content = {
194			"box_id": uid,
195			"namespace": self.namespace,
196			"callback": _find_user_callback,
197			}
198	1		event = KytosEvent(name="kytos.storehouse.retrieve", content=content)
199	1		self.controller.buffers.app.put(event)
200			while True:
201	1		time.sleep(0.1)
202	1		if response:
203	1		break
204	1		return response["answer"], response["code"]
205
206	1		@authenticated
207	1		def _list_user(self, uid):
208			"""List a specific user using Storehouse."""
209	1		answer, code = self._find_user(uid)
210	1		if code == HTTPStatus.OK.value:
211	1		del answer['data']['password']
212	1		return answer, code
213
214	1	View Code Duplication	@authenticated
			0 ignored issues – show Duplication introduced 2021-11-30 19:20 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
215	1		def _list_users(self):
216			"""List all users using Storehouse."""
217	1		response = {}
218
219	1		def _list_users_callback(_event, boxes, error):
220			nonlocal response
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-11-30 19:20 UTC by Report Bug Copy Issue Report The variable `response` does not seem to be defined. Loading history...
221	1		if error:
222			response = {
223			"answer": "Users cannot be listed",
224			"code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
225			}
226			else:
227	1		response = {
228			"answer": {"users": boxes},
229			"code": HTTPStatus.OK.value,
230			}
231
232	1		content = {
233			"namespace": self.namespace,
234			"callback": _list_users_callback,
235			}
236	1		event = KytosEvent(name="kytos.storehouse.list", content=content)
237	1		self.controller.buffers.app.put(event)
238			while True:
239	1		time.sleep(0.1)
240	1		if response:
241	1		break
242	1		return response["answer"], response["code"]
243
244	1		@authenticated
245	1		def _create_user(self):
246			"""Save a user using Storehouse."""
247	1		response = {}
248
249	1		def _create_user_callback(_event, box, error):
250			nonlocal response
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-11-30 19:20 UTC by Report Bug Copy Issue Report The variable `response` does not seem to be defined. Loading history...
251	1		if not box:
252	1		response = {
253			"answer": 'User already exists',
254			"code": HTTPStatus.CONFLICT.value,
255			}
256	1		elif error:
257			response = {
258			"answer": "User has not been created",
259			"code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
260			}
261			else:
262	1		response = {
263			"answer": "User successfully created",
264			"code": HTTPStatus.OK.value,
265			}
266
267	1		req = request.json
268	1		password = req["password"].encode()
269	1		data = {
270			"username": req["username"],
271			"email": req["email"],
272			"password": hashlib.sha512(password).hexdigest(),
273			}
274	1		content = {
275			"namespace": self.namespace,
276			"box_id": data["username"],
277			"data": data,
278			"callback": _create_user_callback,
279			}
280	1		event = KytosEvent(name="kytos.storehouse.create", content=content)
281	1		self.controller.buffers.app.put(event)
282			while True:
283	1		time.sleep(0.1)
284	1		if response:
285	1		break
286	1		return response["answer"], response["code"]
287
288	1	View Code Duplication	@authenticated
			0 ignored issues – show Duplication introduced 2021-11-30 19:20 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
289	1		def _delete_user(self, uid):
290			"""Delete a user using Storehouse."""
291	1		response = {}
292
293	1		def _delete_user_callback(_event, box, error):
294			nonlocal response
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-11-30 19:20 UTC by Report Bug Copy Issue Report The variable `response` does not seem to be defined. Loading history...
295	1		if not box:
296	1		response = {
297			"answer": f'User with uid {uid} not found',
298			"code": HTTPStatus.NOT_FOUND.value
299			}
300	1		elif error:
301			response = {
302			"answer": "User has not been deleted",
303			"code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
304			}
305			else:
306	1		response = {
307			"answer": "User successfully deleted",
308			"code": HTTPStatus.OK.value,
309			}
310
311	1		content = {
312			"box_id": uid,
313			"namespace": self.namespace,
314			"callback": _delete_user_callback,
315			}
316	1		event = KytosEvent(name="kytos.storehouse.delete", content=content)
317	1		self.controller.buffers.app.put(event)
318			while True:
319	1		time.sleep(0.1)
320	1		if response:
321	1		break
322	1		return response["answer"], response["code"]
323
324	1		@authenticated
325	1		def _update_user(self, uid):
326			"""Update user data using Storehouse."""
327	1		response = {}
328
329	1		def _update_user_callback(_event, box, error):
330			nonlocal response
			0 ignored issues – show Comprehensibility Best Practice introduced 2021-11-30 19:20 UTC by Report Bug Copy Issue Report The variable `response` does not seem to be defined. Loading history...
331	1		if not box:
332	1		response = {
333			"answer": f'User with uid {uid} not found',
334			"code": HTTPStatus.NOT_FOUND.value
335			}
336	1		elif error:
337			response = {
338			"answer": "User has not been updated",
339			"code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
340			}
341			else:
342	1		response = {
343			"answer": "User successfully updated",
344			"code": HTTPStatus.OK.value,
345			}
346
347	1		req = request.json
348	1		allowed = ["username", "email", "password"]
349
350	1		data = {}
351	1		for key, value in req.items():
352	1		if key in allowed:
353	1		data[key] = value
354
355	1		content = {
356			"namespace": self.namespace,
357			"box_id": uid,
358			"data": data,
359			"callback": _update_user_callback,
360			}
361	1		event = KytosEvent(name="kytos.storehouse.update", content=content)
362	1		self.controller.buffers.app.put(event)
363			while True:
364	1		time.sleep(0.1)
365	1		if response:
366	1		break
367			return response["answer"], response["code"]
368

kytos-ng / kytos

Pull Request — master (#284)

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

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like