Services

Business logic for the API

server.app.services.api_services.add_custom_ntp_measurement_ip_to_db_measurement(db, server_ip, settings, full_m, from_dn=None)

This method is used when you performed a full measurement on an IP address. If the measurement fails, you will not see why in the full_m.response_error :type db: Session :param db: A connection to the database (we need to query some IDs) :type db: Session :type server_ip: str :param server_ip: The IP address of the server. :type server_ip: str :type settings: AdvancedSettings :param settings: The settings to use. :type settings: AdvancedSettings :type full_m: FullMeasurementIP :param full_m: Full measurement IP object. :type full_m: FullMeasurementIP :type from_dn: Optional[str] :param from_dn: The domain name of this IP address, if available. (it will simplify the

process of searching in the db)

Returns:

nothing

Return type:

None

server.app.services.api_services.add_ntp_measurement(db, result, settings, host, server_ip)

This method adds the result (NTP measurement) into the database. :type db: Session :param db: A connection to the database. :type db: Session :type result: Optional[dict] :param result: the result of the adding NTP measurement :type result: Optional[dict] :type settings: AdvancedSettings :param settings: the settings to use :type settings: AdvancedSettings :type host: str :param host: the host to use (IP or domain name) :type host: Optional[str] :type server_ip: Optional[str] :param server_ip: the IP :type server_ip: Optional[str]

Returns:

A pair of the measurement and its version.

Return type:

Tuple[Optional[NTPv4Measurement | NTPv5Measurement], Optional[str]]

server.app.services.api_services.add_ntp_versions_to_db_measurement(db, server, settings, m, from_dn=None)

This method adds the ntp versions analysis to the database and to the measurement. If this fails, then the ID for this measurement will be None/Null when the status of the overall measurement will be “finished”. :type db: Session :param db: A connection to the database (we need to query some IDs) :type db: Session :type server: str :param server: The server (IP address or domain name) :type server: str :type settings: AdvancedSettings :param settings: The settings to use. :type settings: AdvancedSettings :type m: FullMeasurementDN | FullMeasurementIP :param m: Full measurement IP object. :type m: FullMeasurementDN | FullMeasurementIP :type from_dn: Optional[str] :param from_dn: The domain name of this IP address, if available. :type from_dn: Optional[str]

Returns:

nothing

Return type:

None

server.app.services.api_services.add_ripe_measurement_id_to_db_measurement(db, server, settings, m)

This method perform the RIPE measurement. It marked the ripe_error field if there are any errors. :type db: Session :param db: A connection to the database (we need to query some IDs) :type db: Session :type server: str :param server: The server (IP address or domain name) :type server: str :type settings: AdvancedSettings :param settings: The settings to use. :type settings: AdvancedSettings :type m: FullMeasurementDN | FullMeasurementIP :param m: Full measurement IP object. :type m: FullMeasurementDN | FullMeasurementIP

Returns:

nothing

Return type:

None

server.app.services.api_services.check_and_get_settings(input_settings)

This method takes the parameters that the client inputted. It checks them and it returns the settings. :type input_settings: MeasurementRequest :param input_settings: The parameters that the client inputted. :type input_settings: MeasurementRequest

Returns:

The settings to be used internally in the server. (valid settings)

Return type:

AdvancedSettings

Raises:

InputError – If some settings are invalid.

server.app.services.api_services.check_ripe_measurement_scheduled(measurement_id)

Check if a RIPE Atlas measurement has been fully scheduled.

This function delegates to check_all_measurements_scheduled() to verify that all requested probes have been scheduled for the given RIPE measurement ID.

Parameters:

measurement_id (str) – The ID of the RIPE measurement to check.

Returns:

True if all requested probes are scheduled, False otherwise.

Return type:

bool

Raises:

ValueError – If the RIPE API returns an error or unexpected data.

server.app.services.api_services.check_ripe_settings(settings)

This method checks the values of the RIPE section of the settings. :type settings: AdvancedSettings :param settings: The parameters that the client inputted. :type settings: AdvancedSettings

Returns:

The settings to be used internally in the server. (valid settings)

Return type:

AdvancedSettings

Raises:

InputError – If some settings are invalid.

server.app.services.api_services.check_settings(settings)

This method checks the values in the settings. :type settings: AdvancedSettings :param settings: The parameters that the client inputted. :type settings: AdvancedSettings

Returns:

The settings to be used internally in the server. (valid settings)

Return type:

AdvancedSettings

Raises:

InputError – If some settings are invalid.

server.app.services.api_services.complete_this_measurement_dn(measurement_id, dn_ips, settings)

if the measurement_id is not in the database, then this method does nothing.

Return type:

None

server.app.services.api_services.complete_this_measurement_ip(measurement_id, settings, part_of_dn_measurement=False, from_dn=None)

if the measurement_id is not in the database, then this method does nothing.

Return type:

None

server.app.services.api_services.fetch_historic_data_with_timestamps(server, start, end, session)

Fetches and reconstructs NTP measurements from the database within a specific time range.

Converts the provided human-readable datetime range into NTP-compatible timestamps, queries the database based on whether the server is an IP address or domain name, and reconstructs each result as an NtpMeasurement object.

Parameters:

• server (str) – An IPv4/IPv6 address or domain name string for which measurements should be fetched.

• start (datetime) – The start of the time range (in local or UTC timezone).

• end (datetime) – The end of the time range (in local or UTC timezone).

• session (Session) – The currently active database session.

Returns:

A list of NtpMeasurement objects representing the historical data for the given server within the time window.

Return type:

list[NtpMeasurement]

Notes

• The input datetimes are converted to UTC before processing.

• IP addresses are validated using the is_ip_address() utility function.

• Data is fetched using get_measurements_timestamps_ip or get_measurements_timestamps_dn depending on the server type.

• The PreciseTime wrapper is used to reconstruct accurate timestamps from database fields.

server.app.services.api_services.fetch_ripe_data(measurement_id)

Fetches and formats NTP measurement data from RIPE Atlas.

This function retrieves raw measurement data from the RIPE Atlas API using the given measurement ID, parses it into internal data structures, and formats it into a standardized dictionary format.

Parameters:

measurement_id (str) – The unique ID of the RIPE Atlas measurement to fetch.

Returns:

A list of dictionaries, each representing a formatted NTP measurement.

Return type:

list[dict]

server.app.services.api_services.get_format(measurement, jitter=None, nr_jitter_measurements=8)

Format an NTP measurement object into a dictionary suitable for JSON serialization.

Parameters:

• measurement (NtpMeasurement) – An object representing the NTP measurement result

• jitter (Optional[float]) – Optional jitter value if multiple measurements are performed

• nr_jitter_measurements (int) – The number of measurements used in the jitter calculation

Returns:

A dictionary containing key measurement details like this:

• Server info (ntp version, IP, name, reference IP, reference)

• Timestamps (client sent time, server receive time, server sent time, client receive time)

• Measurement metrics (offset, delay, stratum, precision, reachability)

• Extra details (root delay, last sync time, leap indicator)

Return type:

dict

server.app.services.api_services.get_host_and_server_ip(server, from_dn=None)

This method gets the host and server IP. It is useful because it can work with both cases: domain name and IP address :type server: str :param server: The server name. (ip or dn) :type server: str :type from_dn: Optional[str] :param from_dn: The domain name of this IP address. :type from_dn: Optional[str]

Returns:

The host and server IP.

Return type:

Tuple[str, Optional[str]]

server.app.services.api_services.get_ripe_format(measurement)

Converts a RipeMeasurement object into a standardized dictionary format.

This function extracts relevant information from the provided RipeMeasurement instance—including NTP server info, probe data, timing details, and measurement results—and formats it as a plain dictionary.

Parameters:

measurement (RipeMeasurement) – The parsed measurement object containing NTP and probe data

Returns:

A dictionary containing structured measurement data. Keys include: - NTP Server info (ntp version, ripe measurement id, IP, name, ref id) - Probe data (probe address, probe id in RIPE Atlas, probe location, time to result) - Measurement metrics (stratum, poll, precision, root delay, root dispersion, reachability) - NTP measurement data (rtt, offset, timestamps)

Return type:

dict[str, Any]

server.app.services.api_services.get_server_info_objectv4(db, m_id, server_ip)

Creates the server info object fot this measurement. :type db: Session :param db: A connection to the database (we need to query some IDs). :type db: Session :type m_id: int :param m_id: The measurement (NTPv4 class) id. :type m_id: int :type server_ip: str :param server_ip: The IP address of the server. :type server_ip: Optional[str]

Returns:

nothing

Return type:

None

server.app.services.api_services.get_server_info_objectv5(db, m_id, server_ip)

Creates the server info object fot this measurement. :type db: Session :param db: A connection to the database (we need to query some IDs). :type db: Session :type m_id: int :param m_id: The measurement (NTPv5 class) id. :type m_id: int :type server_ip: str :param server_ip: The IP address of the server. :type server_ip: Optional[str]

Returns:

nothing

Return type:

None

server.app.services.api_services.measure(server, wanted_ip_type, session, client_ip=None, measurement_no=8)

Performs an NTP measurement for a given server (IP or domain name) and stores the result in the database.

This function determines whether the input is an IP address or a domain name, then performs an NTP measurement using the appropriate method. The result is inserted into the database and returned.

Parameters:

• server (str) – A string representing either an IPv4/IPv6 address or a domain name.

• wanted_ip_type (int) – The IP type that we want to measure. Used for domain names.

• session (Session) – The currently active database session.

• client_ip (Optional[str]) – The client IP or None if it was not provided.

• measurement_no (int) – How many extra measurements to perform if the jitter_flag is True.

Returns:

• A list of pairs with a populated NtpMeasurement object if the measurement is successful, and the jitter.

• None if an exception occurs during the measurement process.

Return type:

list[tuple[NtpMeasurement, float, int]] | None

Raises:

DNSError – If the domain name is invalid, or it could not be resolved.

Notes

• If the server string is empty or improperly formatted, this may raise exceptions internally, which are caught and logged to stdout.

• This function modifies persistent state by inserting a measurement into the database.

server.app.services.api_services.override_desired_ip_type_if_input_is_ip(target_server, wanted_ip_type)

If the target server input is IP, then we want to perform its IP type measurements. Only for domain names, “wanted_ip_type” is considered. Otherwise, it is ignored.

Parameters:

• target_server (str) – The server we want to measure (domain name or IP address)

• wanted_ip_type (int) – The IP type the user said they wanted to measure.

Returns:

The IP type of the server in case the server input is IP, otherwise the wanted_ip_type unmodified.

Return type:

int

server.app.services.api_services.perform_ripe_measurement(ntp_server, settings)

Initiate a RIPE Atlas measurement for a given server (IP address or domain name).

This function determines whether the provided server is an IP address or a domain name, and triggers the appropriate RIPE measurement.

Parameters:

• ntp_server (str) – The IP address or domain name of the target NTP server.

• settings (AdvancedSettings) – The settings to use.

Returns:

The RIPE measurement ID. (as a string)

Return type:

str

Raises:

Exception – If the server string is invalid or the measurement failed.

server.app.services.api_services.take_care_of_exception(db, e, measurement_id, ip_measurement)

This method takes care of marking the measurement as failed, if something happened. :type db: Session :param db: A connection to the database (we need to query some IDs). :type db: Session :type e: Exception :param e: An exception instance. :type e: Exception :type measurement_id: int :param measurement_id: The measurement id. :type measurement_id: int :type ip_measurement: bool :param ip_measurement: Whether the measurement was on an IP address. :type ip_measurement: bool

Return type:

None

Methods used for calculating data from the timestamps

class server.app.services.NtpCalculator.NtpCalculator

Bases: object

static calculate_float_time(time)

Converts a PreciseTime object to a float in seconds.

Parameters:

time (PreciseTime) – A single PreciseTime object, representing a single timestamp

Returns:

Time in seconds.

Return type:

float

static calculate_jitter(offsets)

Calculates the jitter of multiple NTP measurements based on their offsets.

Parameters:

offsets (list[float]) – A list of floats representing the offsets to calculate jitter.

Returns:

Jitter in seconds.

Return type:

float

static calculate_offset(timestamps)

Calculates the clock offset between client and server using NTP timestamps. Uses the formula ((t2 - t1) + (t3 - t4)) / 2

Parameters:

timestamps (NtpTimestamps) – A single NTP timestamps object, containing data about the 4 key timestamps

Returns:

Clock offset in seconds.

Return type:

float

static calculate_offset_from_dict(response_dict)

Calculates the clock offset between client and server using NTP timestamps from a dictionary. Uses the formula ((t2 - t1) + (t3 - t4)) / 2

Parameters:

response_dict (dict[str, Any]) – A dictionary with the timestamp values. (names are taken from RIPE results)

Returns:

Clock offset in seconds.

Return type:

float

static calculate_rtt(timestamps)

Calculates round-trip delay between client and server using NTP timestamps. It uses the formula (t4 - t1) - (t3 - t2)

Parameters:

timestamps (NtpTimestamps) – A single NTP timestamps object, containing data about the 4 key timestamps

Returns:

Delay in seconds

Return type:

float

Methods used for validating an NTP measurement

class server.app.services.NtpValidation.NtpValidation

Bases: object

static is_valid(details)

Checks the validity of the details object. According to ntp, the ‘leap’ attribute has only 2 bits and if its value is 3 (11 in binary) then it is invalid.

Parameters:

details (NtpExtraDetails) – The details objects to validate.

Returns:

True if the provided details have a ‘leap’ value different from 3,

False otherwise.

Return type:

bool