Issues in auth.py - New Issues - Inspection of "Update AUTHORS.rst" - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7202e5...a6895b )

by Humberto

created 2020-05-19 19:19 UTC

kytos/core/auth.py (5 issues)

Labels

Severity

Minor 5

"""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__)
TOKEN_EXPIRATION_MINUTES = 180


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 AttributeError
            token = content.split("Bearer ")[1]
            jwt.decode(token, key=Auth.get_jwt_secret())
        except (
            AttributeError,
            IndexError,
            jwt.ExpiredSignature,
            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."""

    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"
        if self.controller.options.create_superuser is True:
            self._create_superuser()

    @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='HS256',
        )

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

kytos / kytos

Push — master ( 7202e5...a6895b )

kytos/core/auth.py (5 issues)

Labels

Severity

Introduced By

Duplication Side-by-Side

Filter issues like