17220817776

Committed 25 Aug 2025 09:09PM UTC coverage: 80.753% (+1.7%) from 79.016%

Build # 17220817776

Build Type

Pull #96

github

Committed by

actions-user

Commit Message

Apply automatic documentation changes

Pull Request Pull Request #96: Feat/add additional-data column

Run Details

80 of 85 new or added lines in 5 files covered. (94.12%)

5 existing lines in 2 files now uncovered.

558 of 691 relevant lines covered (80.75%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.46

/src/audit/utils/validate_utils.py

from datetime import datetime
from fastapi import HTTPException
from starlette.status import HTTP_400_BAD_REQUEST
import time

from .. import logger
from ..config import config


def handle_timestamp(data):
    """
    If the timestamp is omitted from the request body, we use the current date
    and time. In most cases, services should NOT provide a timestamp when
    creating audit logs. The timestamp is only accepted in log creation
    requests to allow populating the audit database with historical data, for
    example by parsing historical logs from before the Audit Service was
    deployed to a Data Commons.
    """
    if "timestamp" not in data:
        return
    if data["timestamp"]:
        # we take a timestamp as input, but store a datetime in the database
        try:
            data["timestamp"] = datetime.fromtimestamp(data["timestamp"])
        except Exception as e:
            raise HTTPException(
                HTTP_400_BAD_REQUEST,
                f"Invalid timestamp '{data['timestamp']}'",
            )
    else:
        # when hitting the API endpoint, the "timestamp" key always exists
        # because it's defined in `CreateLogInput`. It is automatically added
        # to rows without timestamp, and is defaulted to None.
        # Setting it to the current date and time.
        data["timestamp"] = datetime.now()


def validate_presigned_url_log(data):
    logger.debug(f"Creating `presigned_url` audit log. Received body: {data}")

    allowed_actions = ["download", "upload"]
    # `action` is a required field", but that's checked during the DB insert
    if "action" in data and data["action"] not in allowed_actions:
        raise HTTPException(
            HTTP_400_BAD_REQUEST,
            f"Action '{data['action']}' is not allowed ({allowed_actions})",
        )

    handle_timestamp(data)


def validate_login_log(data):
    logger.debug(f"Creating `login` audit log. Received body: {data}")
    handle_timestamp(data)
    if data.get("ip") is None:
        logger.warning("login log received in deprecated legacy format (missing ip)")


def validate_and_normalize_times(start, stop):
    """
    Validate the `start` and `stop` parameters, raise exceptions if the
    validation fails, and return:
    - start: if parameter `start` was None, override with default
    - start_date: `start` as datetime
    - stop: if parameter `stop` was None, override with default
    - stop_date: `stop` as datetime
    """
    # don't just overwrite `stop` with the current timestamp, because the
    # `stop` param is exclusive and when we don't specify it, we want to
    # be able to query logs we just created
    effective_stop = stop or int(time.time())

    timebox_max_seconds = None
    if config["QUERY_TIMEBOX_MAX_DAYS"]:
        timebox_max_seconds = config["QUERY_TIMEBOX_MAX_DAYS"] * 86400

        # if the query is time-boxed and `stop` was not specified,
        # set `stop` to the newest allowed timestamp
        if start and not stop:
            stop = start + timebox_max_seconds
            effective_stop = stop

        # if the query is time-boxed and `start` was not specified,
        # set `start` to the oldest allowed timestamp
        if not start:
            start = max(effective_stop - timebox_max_seconds, 0)

    start_date = None
    stop_date = None
    try:
        if start:
            start_date = datetime.fromtimestamp(start)
        if stop:
            stop_date = datetime.fromtimestamp(stop)
    except Exception as e:
        msg = f"Unable to convert timestamps '{start}' and/or '{stop}' to datetimes"
        logger.error(f"{msg}:\n{e}")
        raise HTTPException(HTTP_400_BAD_REQUEST, msg)

    if start and stop and start > stop:
        raise HTTPException(
            HTTP_400_BAD_REQUEST,
            f"The start timestamp '{start}' ({start_date}) should be before the stop timestamp '{stop}' ({stop_date})",
        )

    if timebox_max_seconds and effective_stop - start > timebox_max_seconds:
        raise HTTPException(
            HTTP_400_BAD_REQUEST,
            f"The difference between the start timestamp '{start}' ({start_date}) and the stop timestamp '{stop}' ({stop_date}) is greater than the configured maximum of {config['QUERY_TIMEBOX_MAX_DAYS']} days",
        )

    return start, start_date, stop, stop_date

1	from datetime import datetime	1✔
2	from fastapi import HTTPException	1✔
3	from starlette.status import HTTP_400_BAD_REQUEST	1✔
4	import time	1✔
5
6	from .. import logger	1✔
7	from ..config import config	1✔
8
9
10	def handle_timestamp(data):	1✔
11	"""
12	If the timestamp is omitted from the request body, we use the current date
13	and time. In most cases, services should NOT provide a timestamp when
14	creating audit logs. The timestamp is only accepted in log creation
15	requests to allow populating the audit database with historical data, for
16	example by parsing historical logs from before the Audit Service was
17	deployed to a Data Commons.
18	"""
19	if "timestamp" not in data:	1✔
20	return	×
21	if data["timestamp"]:	1✔
22	# we take a timestamp as input, but store a datetime in the database
23	try:	1✔
24	data["timestamp"] = datetime.fromtimestamp(data["timestamp"])	1✔
25	except Exception as e:	1✔
26	raise HTTPException(	1✔
27	HTTP_400_BAD_REQUEST,
28	f"Invalid timestamp '{data['timestamp']}'",
29	)
30	else:
31	# when hitting the API endpoint, the "timestamp" key always exists
32	# because it's defined in `CreateLogInput`. It is automatically added
33	# to rows without timestamp, and is defaulted to None.
34	# Setting it to the current date and time.
35	data["timestamp"] = datetime.now()	1✔
36
37
38	def validate_presigned_url_log(data):	1✔
39	logger.debug(f"Creating `presigned_url` audit log. Received body: {data}")	1✔
40
41	allowed_actions = ["download", "upload"]	1✔
42	# `action` is a required field", but that's checked during the DB insert
43	if "action" in data and data["action"] not in allowed_actions:	1✔
44	raise HTTPException(	1✔
45	HTTP_400_BAD_REQUEST,
46	f"Action '{data['action']}' is not allowed ({allowed_actions})",
47	)
48
49	handle_timestamp(data)	1✔
50
51
52	def validate_login_log(data):	1✔
53	logger.debug(f"Creating `login` audit log. Received body: {data}")	1✔
54	handle_timestamp(data)	1✔
55	if data.get("ip") is None:	1✔
56	logger.warning("login log received in deprecated legacy format (missing ip)")	1✔
57
58
59	def validate_and_normalize_times(start, stop):	1✔
60	"""
61	Validate the `start` and `stop` parameters, raise exceptions if the
62	validation fails, and return:
63	- start: if parameter `start` was None, override with default
64	- start_date: `start` as datetime
65	- stop: if parameter `stop` was None, override with default
66	- stop_date: `stop` as datetime
67	"""
68	# don't just overwrite `stop` with the current timestamp, because the
69	# `stop` param is exclusive and when we don't specify it, we want to
70	# be able to query logs we just created
71	effective_stop = stop or int(time.time())	1✔
72
73	timebox_max_seconds = None	1✔
74	if config["QUERY_TIMEBOX_MAX_DAYS"]:	1✔
75	timebox_max_seconds = config["QUERY_TIMEBOX_MAX_DAYS"] * 86400	1✔
76
77	# if the query is time-boxed and `stop` was not specified,
78	# set `stop` to the newest allowed timestamp
79	if start and not stop:	1✔
80	stop = start + timebox_max_seconds	1✔
81	effective_stop = stop	1✔
82
83	# if the query is time-boxed and `start` was not specified,
84	# set `start` to the oldest allowed timestamp
85	if not start:	1✔
86	start = max(effective_stop - timebox_max_seconds, 0)	1✔
87
88	start_date = None	1✔
89	stop_date = None	1✔
90	try:	1✔
91	if start:	1✔
92	start_date = datetime.fromtimestamp(start)	1✔
93	if stop:	1✔
94	stop_date = datetime.fromtimestamp(stop)	1✔
95	except Exception as e:	×
96	msg = f"Unable to convert timestamps '{start}' and/or '{stop}' to datetimes"	×
UNCOV 97	logger.error(f"{msg}:\n{e}")	×
UNCOV 98	raise HTTPException(HTTP_400_BAD_REQUEST, msg)	×
99
100	if start and stop and start > stop:	1✔
UNCOV 101	raise HTTPException(	×
102	HTTP_400_BAD_REQUEST,
103	f"The start timestamp '{start}' ({start_date}) should be before the stop timestamp '{stop}' ({stop_date})",
104	)
105
106	if timebox_max_seconds and effective_stop - start > timebox_max_seconds:	1✔
107	raise HTTPException(	1✔
108	HTTP_400_BAD_REQUEST,
109	f"The difference between the start timestamp '{start}' ({start_date}) and the stop timestamp '{stop}' ({stop_date}) is greater than the configured maximum of {config['QUERY_TIMEBOX_MAX_DAYS']} days",
110	)
111
112	return start, start_date, stop, stop_date	1✔

uc-cdis / audit-service / 17220817776

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous