core.useractivity.admin_control() - Code Metrics - Inspection of "Merge branch 'develop'" - MyEMS/myems - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( cb176a...4b8481 )

unknown

created 2025-12-17 14:15 UTC

core.useractivity.admin_control() C

↳ Parent: core.useractivity

Complexity

Conditions

Size

Total Lines	67
Code Lines	38

Duplication

Lines	53
Ratio	79.1 %

Importance

Changes

Metric	Value
eloc	38
dl	53
loc	67
rs	5.9999
c	0
b	0
f	0
cc	10
nop	1

How to fix Long Method Complexity

import os
import uuid
from datetime import datetime
from functools import wraps
import falcon
import mysql.connector
import simplejson as json
from gunicorn.http.body import Body
import config


def admin_control(req):

    """
    Check administrator privilege in request headers to protect resources from invalid access.

    This function validates that the request contains valid administrator credentials
    by checking the USER-UUID and TOKEN headers against the user database sessions.
    It ensures that only authenticated administrators can access protected resources.

    Args:
        req: HTTP request object containing headers with USER-UUID and TOKEN

    Raises:
        falcon.HTTPError: If invalid credentials or expired session

    Returns:
        None: If validation passes
    """
    # Validate USER-UUID header
    if 'USER-UUID' not in req.headers or \
            not isinstance(req.headers['USER-UUID'], str) or \
            len(str.strip(req.headers['USER-UUID'])) == 0:
        raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                               description='API.INVALID_USER_UUID')
    admin_user_uuid = str.strip(req.headers['USER-UUID'])

    # Validate TOKEN header
    if 'TOKEN' not in req.headers or \
            not isinstance(req.headers['TOKEN'], str) or \
            len(str.strip(req.headers['TOKEN'])) == 0:
        raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                               description='API.INVALID_TOKEN')
    admin_token = str.strip(req.headers['TOKEN'])

    # Check administrator session in user database
    cnx = mysql.connector.connect(**config.myems_user_db)
    cursor = cnx.cursor()
    query = (" SELECT utc_expires "
             " FROM tbl_sessions "
             " WHERE user_uuid = %s AND token = %s")
    cursor.execute(query, (admin_user_uuid, admin_token,))
    row = cursor.fetchone()

    if row is None:
        cursor.close()
        cnx.close()
        raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
                               description='API.ADMINISTRATOR_SESSION_NOT_FOUND')
    else:
        utc_expires = row[0]
        # Check if session has expired
        if datetime.utcnow() > utc_expires:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
                                   description='API.ADMINISTRATOR_SESSION_TIMEOUT')
    # check administrator privilege
    query = (" SELECT name "
             " FROM tbl_users "
             " WHERE uuid = %s AND is_admin = 1 AND is_read_only = 0 ")
    cursor.execute(query, (admin_user_uuid,))
    row = cursor.fetchone()
    cursor.close()
    cnx.close()
    if row is None:
        raise falcon.HTTPError(status=falcon.HTTP_400,
                               title='API.BAD_REQUEST',
                               description='API.INVALID_PRIVILEGE')


def access_control(req):

    """
        Check user privilege in request headers to protect resources from invalid access
        :param req: HTTP request
        :return: HTTPError if invalid else None
        """
    if 'USER-UUID' not in req.headers or \
            not isinstance(req.headers['USER-UUID'], str) or \
            len(str.strip(req.headers['USER-UUID'])) == 0:
        raise falcon.HTTPError(status=falcon.HTTP_400,
                               title='API.BAD_REQUEST',
                               description='API.INVALID_USER_UUID')
    user_uuid = str.strip(req.headers['USER-UUID'])

    if 'TOKEN' not in req.headers or \
            not isinstance(req.headers['TOKEN'], str) or \
            len(str.strip(req.headers['TOKEN'])) == 0:
        raise falcon.HTTPError(status=falcon.HTTP_400,
                               title='API.BAD_REQUEST',
                               description='API.INVALID_TOKEN')
    ordinary_token = str.strip(req.headers['TOKEN'])

    # Check user session
    cnx = mysql.connector.connect(**config.myems_user_db)
    cursor = cnx.cursor()
    query = (" SELECT utc_expires "
             " FROM tbl_sessions "
             " WHERE user_uuid = %s AND token = %s")
    cursor.execute(query, (user_uuid, ordinary_token,))
    row = cursor.fetchone()

    if row is None:
        cursor.close()
        cnx.close()
        raise falcon.HTTPError(status=falcon.HTTP_404,
                               title='API.NOT_FOUND',
                               description='API.USER_SESSION_NOT_FOUND')
    else:
        utc_expires = row[0]
        if datetime.utcnow() > utc_expires:
            cursor.close()
            cnx.close()
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.USER_SESSION_TIMEOUT')
    # todo: check user privilege
    query = (" SELECT name "
             " FROM tbl_users "
             " WHERE uuid = %s ")
    cursor.execute(query, (user_uuid,))
    row = cursor.fetchone()
    cursor.close()
    cnx.close()
    if row is None:
        raise falcon.HTTPError(status=falcon.HTTP_400,
                               title='API.BAD_REQUEST',
                               description='API.INVALID_PRIVILEGE')


def api_key_control(req):
    """
        Check API privilege in request headers to protect resources from invalid access
        :param req: HTTP request
        :return: HTTPError if invalid else None
    """
    api_key = str.strip(req.headers['API-KEY'])
    cnx = mysql.connector.connect(**config.myems_user_db)
    cursor = cnx.cursor()
    query = (" SELECT expires_datetime_utc "
             " FROM tbl_api_keys "
             " WHERE token = %s ")
    cursor.execute(query, (api_key,))
    row = cursor.fetchone()
    cursor.close()
    cnx.close()
    if row is None:
        raise falcon.HTTPError(status=falcon.HTTP_404,
                               title='API.NOT_FOUND',
                               description='API.API_KEY_NOT_FOUND')
    else:
        expires_datetime_utc = row[0]
        if datetime.utcnow() > expires_datetime_utc:
            raise falcon.HTTPError(status=falcon.HTTP_400,
                                   title='API.BAD_REQUEST',
                                   description='API.API_KEY_HAS_EXPIRED')


def write_log(user_uuid, request_method, resource_type, resource_id, request_body):
    """
    :param user_uuid: user_uuid
    :param request_method: 'POST', 'PUT', 'DELETE'
    :param resource_type: class_name
    :param resource_id: int
    :param request_body: json in raw string
    """
    cnx = None
    cursor = None
    try:
        cnx = mysql.connector.connect(**config.myems_user_db)
        cursor = cnx.cursor()
        add_row = (" INSERT INTO tbl_logs "
                   "    (user_uuid, request_datetime_utc, request_method, resource_type, resource_id, request_body) "
                   " VALUES (%s, %s, %s, %s, %s , %s) ")
        cursor.execute(add_row, (user_uuid,
                                 datetime.utcnow(),
                                 request_method,
                                 resource_type,
                                 resource_id if resource_id else None,
                                 request_body if request_body else None,
                                 ))
        cnx.commit()
    except InterfaceError as e:

        print("Failed to connect request")
    except OperationalError as e:      

        print("Failed to SQL operate request")
    except ProgrammingError as e:

        print("Failed to SQL request")
    except DataError as e:

        print("Failed to SQL Data request")
    except Exception as e:
        print('write_log:' + str(e))
    finally:
        if cursor:
            cursor.close()
        if cnx:
            cnx.close()


def user_logger(func):
    """
    Decorator for logging user activities
    :param func: the decorated function
    :return: the decorator
    """
    @wraps(func)
    def logger(*args, **kwargs):
        qualified_name = func.__qualname__
        class_name = qualified_name.split(".")[0]
        func_name = qualified_name.split(".")[1]

        if func_name not in ("on_post", "on_put", "on_delete"):
            # do not log for other HTTP Methods
            func(*args, **kwargs)
            return
        req, resp = args
        headers = req.headers
        if headers is not None and 'USER-UUID' in headers.keys():
            user_uuid = headers['USER-UUID']
        else:
            # todo: deal with requests with NULL user_uuid
            print('user_logger: USER-UUID is NULL')
            # do not log for NULL user_uuid
            func(*args, **kwargs)
            return

        if func_name == "on_post":
            try:
                file_name = str(uuid.uuid4())
                with open(file_name, "wb") as fw:
                    reads = req.stream.read()
                    fw.write(reads)
                raw_json = reads.decode('utf-8')
                with open(file_name, "rb") as fr:
                    req.stream = Body(fr)
                    func(*args, **kwargs)
                    write_log(user_uuid=user_uuid, request_method='POST', resource_type=class_name,
                              resource_id=kwargs.get('id_'), request_body=raw_json)
                os.remove(file_name)
            except OSError as e:
                print("Failed to stream request")
            except UnicodeDecodeError as e:
                print("Failed to decode request")
            except Exception as e:
                if isinstance(e, falcon.HTTPError):
                    raise e
                else:
                    print('user_logger:' + str(e))
            return
        elif func_name == "on_put":
            try:
                file_name = str(uuid.uuid4())

                with open(file_name, "wb") as fw:
                    reads = req.stream.read()
                    fw.write(reads)
                raw_json = reads.decode('utf-8')
                with open(file_name, "rb") as fr:
                    req.stream = Body(fr)
                    func(*args, **kwargs)
                    write_log(user_uuid=user_uuid, request_method='PUT', resource_type=class_name,
                              resource_id=kwargs.get('id_'), request_body=raw_json)
                os.remove(file_name)
            except OSError as e:
                print("Failed to stream request")
            except UnicodeDecodeError as e:
                print("Failed to decode request")
            except Exception as e:
                if isinstance(e, falcon.HTTPError):
                    raise e
                else:
                    print('user_logger:' + str(e))

            return
        elif func_name == "on_delete":
            try:
                func(*args, **kwargs)
                write_log(user_uuid=user_uuid, request_method="DELETE", resource_type=class_name,
                          resource_id=kwargs.get('id_'), request_body=json.dumps(kwargs))
            except (TypeError, ValueError) as e:
                print("Failed to decode JSON")
            except Exception as e:
                if isinstance(e, falcon.HTTPError):
                    raise e
                else:
                    print('user_logger:' + str(e))
            return

    return logger


1		import os
2		import uuid
3		from datetime import datetime
4		from functools import wraps
5		import falcon
6		import mysql.connector
7		import simplejson as json
8		from gunicorn.http.body import Body
9		import config
10
11
12	View Code Duplication	def admin_control(req):
		0 ignored issues – show Duplication introduced 2023-06-18 11:58 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
13		"""
14		Check administrator privilege in request headers to protect resources from invalid access.
15
16		This function validates that the request contains valid administrator credentials
17		by checking the USER-UUID and TOKEN headers against the user database sessions.
18		It ensures that only authenticated administrators can access protected resources.
19
20		Args:
21		req: HTTP request object containing headers with USER-UUID and TOKEN
22
23		Raises:
24		falcon.HTTPError: If invalid credentials or expired session
25
26		Returns:
27		None: If validation passes
28		"""
29		# Validate USER-UUID header
30		if 'USER-UUID' not in req.headers or \
31		not isinstance(req.headers['USER-UUID'], str) or \
32		len(str.strip(req.headers['USER-UUID'])) == 0:
33		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
34		description='API.INVALID_USER_UUID')
35		admin_user_uuid = str.strip(req.headers['USER-UUID'])
36
37		# Validate TOKEN header
38		if 'TOKEN' not in req.headers or \
39		not isinstance(req.headers['TOKEN'], str) or \
40		len(str.strip(req.headers['TOKEN'])) == 0:
41		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
42		description='API.INVALID_TOKEN')
43		admin_token = str.strip(req.headers['TOKEN'])
44
45		# Check administrator session in user database
46		cnx = mysql.connector.connect(**config.myems_user_db)
47		cursor = cnx.cursor()
48		query = (" SELECT utc_expires "
49		" FROM tbl_sessions "
50		" WHERE user_uuid = %s AND token = %s")
51		cursor.execute(query, (admin_user_uuid, admin_token,))
52		row = cursor.fetchone()
53
54		if row is None:
55		cursor.close()
56		cnx.close()
57		raise falcon.HTTPError(status=falcon.HTTP_404, title='API.NOT_FOUND',
58		description='API.ADMINISTRATOR_SESSION_NOT_FOUND')
59		else:
60		utc_expires = row[0]
61		# Check if session has expired
62		if datetime.utcnow() > utc_expires:
63		cursor.close()
64		cnx.close()
65		raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST',
66		description='API.ADMINISTRATOR_SESSION_TIMEOUT')
67		# check administrator privilege
68		query = (" SELECT name "
69		" FROM tbl_users "
70		" WHERE uuid = %s AND is_admin = 1 AND is_read_only = 0 ")
71		cursor.execute(query, (admin_user_uuid,))
72		row = cursor.fetchone()
73		cursor.close()
74		cnx.close()
75		if row is None:
76		raise falcon.HTTPError(status=falcon.HTTP_400,
77		title='API.BAD_REQUEST',
78		description='API.INVALID_PRIVILEGE')
79
80
81	View Code Duplication	def access_control(req):
		0 ignored issues – show Duplication introduced 2023-06-18 11:58 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
82		"""
83		Check user privilege in request headers to protect resources from invalid access
84		:param req: HTTP request
85		:return: HTTPError if invalid else None
86		"""
87		if 'USER-UUID' not in req.headers or \
88		not isinstance(req.headers['USER-UUID'], str) or \
89		len(str.strip(req.headers['USER-UUID'])) == 0:
90		raise falcon.HTTPError(status=falcon.HTTP_400,
91		title='API.BAD_REQUEST',
92		description='API.INVALID_USER_UUID')
93		user_uuid = str.strip(req.headers['USER-UUID'])
94
95		if 'TOKEN' not in req.headers or \
96		not isinstance(req.headers['TOKEN'], str) or \
97		len(str.strip(req.headers['TOKEN'])) == 0:
98		raise falcon.HTTPError(status=falcon.HTTP_400,
99		title='API.BAD_REQUEST',
100		description='API.INVALID_TOKEN')
101		ordinary_token = str.strip(req.headers['TOKEN'])
102
103		# Check user session
104		cnx = mysql.connector.connect(**config.myems_user_db)
105		cursor = cnx.cursor()
106		query = (" SELECT utc_expires "
107		" FROM tbl_sessions "
108		" WHERE user_uuid = %s AND token = %s")
109		cursor.execute(query, (user_uuid, ordinary_token,))
110		row = cursor.fetchone()
111
112		if row is None:
113		cursor.close()
114		cnx.close()
115		raise falcon.HTTPError(status=falcon.HTTP_404,
116		title='API.NOT_FOUND',
117		description='API.USER_SESSION_NOT_FOUND')
118		else:
119		utc_expires = row[0]
120		if datetime.utcnow() > utc_expires:
121		cursor.close()
122		cnx.close()
123		raise falcon.HTTPError(status=falcon.HTTP_400,
124		title='API.BAD_REQUEST',
125		description='API.USER_SESSION_TIMEOUT')
126		# todo: check user privilege
127		query = (" SELECT name "
128		" FROM tbl_users "
129		" WHERE uuid = %s ")
130		cursor.execute(query, (user_uuid,))
131		row = cursor.fetchone()
132		cursor.close()
133		cnx.close()
134		if row is None:
135		raise falcon.HTTPError(status=falcon.HTTP_400,
136		title='API.BAD_REQUEST',
137		description='API.INVALID_PRIVILEGE')
138
139
140		def api_key_control(req):
141		"""
142		Check API privilege in request headers to protect resources from invalid access
143		:param req: HTTP request
144		:return: HTTPError if invalid else None
145		"""
146		api_key = str.strip(req.headers['API-KEY'])
147		cnx = mysql.connector.connect(**config.myems_user_db)
148		cursor = cnx.cursor()
149		query = (" SELECT expires_datetime_utc "
150		" FROM tbl_api_keys "
151		" WHERE token = %s ")
152		cursor.execute(query, (api_key,))
153		row = cursor.fetchone()
154		cursor.close()
155		cnx.close()
156		if row is None:
157		raise falcon.HTTPError(status=falcon.HTTP_404,
158		title='API.NOT_FOUND',
159		description='API.API_KEY_NOT_FOUND')
160		else:
161		expires_datetime_utc = row[0]
162		if datetime.utcnow() > expires_datetime_utc:
163		raise falcon.HTTPError(status=falcon.HTTP_400,
164		title='API.BAD_REQUEST',
165		description='API.API_KEY_HAS_EXPIRED')
166
167
168		def write_log(user_uuid, request_method, resource_type, resource_id, request_body):
169		"""
170		:param user_uuid: user_uuid
171		:param request_method: 'POST', 'PUT', 'DELETE'
172		:param resource_type: class_name
173		:param resource_id: int
174		:param request_body: json in raw string
175		"""
176		cnx = None
177		cursor = None
178		try:
179		cnx = mysql.connector.connect(**config.myems_user_db)
180		cursor = cnx.cursor()
181		add_row = (" INSERT INTO tbl_logs "
182		" (user_uuid, request_datetime_utc, request_method, resource_type, resource_id, request_body) "
183		" VALUES (%s, %s, %s, %s, %s , %s) ")
184		cursor.execute(add_row, (user_uuid,
185		datetime.utcnow(),
186		request_method,
187		resource_type,
188		resource_id if resource_id else None,
189		request_body if request_body else None,
190		))
191		cnx.commit()
192		except InterfaceError as e:
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-12-17 14:28 UTC by Report Bug Copy Issue Report The variable `InterfaceError` does not seem to be defined. Loading history...
193		print("Failed to connect request")
194		except OperationalError as e:
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-12-17 14:28 UTC by Report Bug Copy Issue Report The variable `OperationalError` does not seem to be defined. Loading history...
195		print("Failed to SQL operate request")
196		except ProgrammingError as e:
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-12-17 14:28 UTC by Report Bug Copy Issue Report The variable `ProgrammingError` does not seem to be defined. Loading history...
197		print("Failed to SQL request")
198		except DataError as e:
		0 ignored issues – show Comprehensibility Best Practice introduced 2025-12-17 14:28 UTC by Report Bug Copy Issue Report The variable `DataError` does not seem to be defined. Loading history...
199		print("Failed to SQL Data request")
200		except Exception as e:
201		print('write_log:' + str(e))
202		finally:
203		if cursor:
204		cursor.close()
205		if cnx:
206		cnx.close()
207
208
209		def user_logger(func):
210		"""
211		Decorator for logging user activities
212		:param func: the decorated function
213		:return: the decorator
214		"""
215		@wraps(func)
216		def logger(args, *kwargs):
217		qualified_name = func.__qualname__
218		class_name = qualified_name.split(".")[0]
219		func_name = qualified_name.split(".")[1]
220
221		if func_name not in ("on_post", "on_put", "on_delete"):
222		# do not log for other HTTP Methods
223		func(args, *kwargs)
224		return
225		req, resp = args
226		headers = req.headers
227		if headers is not None and 'USER-UUID' in headers.keys():
228		user_uuid = headers['USER-UUID']
229		else:
230		# todo: deal with requests with NULL user_uuid
231		print('user_logger: USER-UUID is NULL')
232		# do not log for NULL user_uuid
233		func(args, *kwargs)
234		return
235
236		if func_name == "on_post":
237		try:
238		file_name = str(uuid.uuid4())
239		with open(file_name, "wb") as fw:
240		reads = req.stream.read()
241		fw.write(reads)
242		raw_json = reads.decode('utf-8')
243		with open(file_name, "rb") as fr:
244		req.stream = Body(fr)
245		func(args, *kwargs)
246		write_log(user_uuid=user_uuid, request_method='POST', resource_type=class_name,
247		resource_id=kwargs.get('id_'), request_body=raw_json)
248		os.remove(file_name)
249		except OSError as e:
250		print("Failed to stream request")
251		except UnicodeDecodeError as e:
252		print("Failed to decode request")
253		except Exception as e:
254		if isinstance(e, falcon.HTTPError):
255		raise e
256		else:
257		print('user_logger:' + str(e))
258		return
259		elif func_name == "on_put":
260		try:
261		file_name = str(uuid.uuid4())
262
263		with open(file_name, "wb") as fw:
264		reads = req.stream.read()
265		fw.write(reads)
266		raw_json = reads.decode('utf-8')
267		with open(file_name, "rb") as fr:
268		req.stream = Body(fr)
269		func(args, *kwargs)
270		write_log(user_uuid=user_uuid, request_method='PUT', resource_type=class_name,
271		resource_id=kwargs.get('id_'), request_body=raw_json)
272		os.remove(file_name)
273		except OSError as e:
274		print("Failed to stream request")
275		except UnicodeDecodeError as e:
276		print("Failed to decode request")
277		except Exception as e:
278		if isinstance(e, falcon.HTTPError):
279		raise e
280		else:
281		print('user_logger:' + str(e))
282
283		return
284		elif func_name == "on_delete":
285		try:
286		func(args, *kwargs)
287		write_log(user_uuid=user_uuid, request_method="DELETE", resource_type=class_name,
288		resource_id=kwargs.get('id_'), request_body=json.dumps(kwargs))
289		except (TypeError, ValueError) as e:
290		print("Failed to decode JSON")
291		except Exception as e:
292		if isinstance(e, falcon.HTTPError):
293		raise e
294		else:
295		print('user_logger:' + str(e))
296		return
297
298		return logger
299

MyEMS / myems

Push — master ( cb176a...4b8481 )

core.useractivity.admin_control() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like