Utilities

Methods used to perform calculations

server.app.utils.calculations.calculate_haversine_distance(lat1, lon1, lat2, lon2)

It calculates the haversine distance between two points using this formula: d = 2 * R * asin(sqrt(a)) where R is Earth’s radius in kilometers and ‘a’ is a value calculated using the latitude and longitude difference of the two points.

Parameters:

• lat1 (float) – The latitude of the first point.

• lon1 (float) – The longitude of the first point.

• lat2 (float) – The latitude of the second point.

• lon2 (float) – The longitude of the second point.

Returns:

The haversine distance between the two points. (in kilometers)

Return type:

float

server.app.utils.calculations.calculate_jitter_from_measurements(session, initial_measurement, no_measurements=8)

Calculates the NTP jitter based on a set of previous measurements and one initial reference measurement.

This function computes the jitter by calculating the standard deviation of the offsets from a given initial measurement and a number of most recent measurements from the same NTP server.

Parameters:

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

• initial_measurement (NtpMeasurement) – The reference measurement not already stored in the database, used as the baseline for offset comparison.

• no_measurements (int) – The number of recent historical measurements to fetch from the database for jitter calculation.

Returns:

• float: The calculated jitter in seconds.

• int: The actual number of historical measurements used for the calculation.

Return type:

tuple[float, int]

server.app.utils.calculations.convert_float_to_precise_time(value)

Converts a float value to a PreciseTime object.

Parameters:

value (float) – The float value to convert.

Returns:

A PreciseTime object.

Return type:

PreciseTime

server.app.utils.calculations.get_non_responding_ntp_measurement(server_ip_str, server_name, ntp_version=4)

Construct a default NTP measurement result representing a non-responding NTP server. This function is used when an NTP server fails to respond. It returns a synthetic NtpMeasurement object with placeholder values (e.g., -1) to indicate that no real measurement was completed. This is used to mark the server as non-responding on the map.

Parameters:

• server_ip_str (str) – The IP address of the NTP server that failed to respond.

• server_name (Optional[str]) – The hostname of the NTP server, if available.

• ntp_version (int) – The version of the NTP protocol to report. (default is based on system config)

Returns:

An NtpMeasurement object filled with placeholder values indicating failure.

Return type:

NtpMeasurement

Notes

• The offset, rtt, stratum, and other time-related fields are set to -1 or equivalent.

• The vantage_point_ip is determined from the local server. If not resolvable, it defaults to 0.0.0.0.

• The location and reference information is generated using available utility functions based on IP.

server.app.utils.calculations.human_date_to_ntp_precise_time(dt)

Converts a UTC datetime object to a PreciseTime object in NTP time.

Parameters:

dt (datetime) – A timezone-aware datetime object in UTC.

Returns:

The corresponding NTP time.

Return type:

PreciseTime

server.app.utils.calculations.ntp_precise_time_to_human_date(t)

Converts a PreciseTime object to a human-readable time string in UTC. (ex:’2025-05-05 14:30:15.123456 UTC’) We need to shift from ntp time to unix time so we need to subtract all the seconds from 1900 to 1970.

Parameters:

t (PreciseTime) – The PreciseTime object.

Returns:

The date in UTC format or empty, depending on whether the PreciseTime object could be converted to UTC.

Return type:

str

Methods used for converting a domain name to an ip

server.app.utils.domain_name_to_ip.domain_name_to_ip_close_to_client(domain_name, client_ip, wanted_ip_type, resolvers=['8.8.8.8', '1.1.1.1', '2001:4860:4860::8888'], depth=0, max_depth=2)

This method tries to obtain the ip addresses of the domain name from some popular DNS servers (resolvers) that have (or may have) the ability to get an IP close to the client. It uses EDNS queries to get the IPs and in case the queries return a domain name, this method recursively tries to solve them. It uses “depth” and “max_depth” to prevent infinite loops in redirecting.

It is important to note that multiple servers may share the same IP address. So, some countries may use the same IP for the same domain name. You can check this using https://www.whatsmydns.net/. This also provides insights in the cases where we receive CNAME responses.

If the name is not a domain name, it will return an empty list. If the EDNS query does not return a CNAME, depth and max_depth would not be used.

Parameters:

• domain_name (str) – The domain name.

• client_ip (str) – The client IP.

• wanted_ip_type (int) – The IP type of the resulting IPs that we want.

• resolvers (list) – A list of popular DNS resolvers that are ECS-capable.

• depth (int) – The depth of the EDNS query if it returns a CNAME. (It is recommended to set this to 0.)

• max_depth (int) – The maximum depth of the EDNS query. (It is recommended to set this to 2 or 3 to prevent long delay.)

Returns:

A list of IPs of the domain name or None if the domain name is not valid.

Return type:

Optional[list[str]]

Raises:

Exception – If the client IP is invalid.

server.app.utils.domain_name_to_ip.domain_name_to_ip_default(domain_name)

It uses the DNS of this server to obtain the ip addresses of the domain name. This method is useful if you want IPs close to this server, or you do not care about the location of the IPs.

Parameters:

domain_name (str) – The NTP server domain name.

Returns:

A list of IPs of the domain name or None if the domain name is not valid.

Return type:

Optional[list[str]]

server.app.utils.domain_name_to_ip.domain_name_to_ip_list(ntp_server_domain_name, client_ip, wanted_ip_type)

This method handles the case when client IP is None and uses our server as the default IP. It will return the list of IP addresses that are close to the client.

Parameters:

• ntp_server_domain_name (str) – NTP server domain name.

• client_ip (Optional[str]) – Client IP address.

• wanted_ip_type (int) – The IP type of the resulting IPs that we want. (IPv4 or IPv6).

Returns:

List of IP addresses that are close to the client or to the server if client IP is None

Return type:

list[str]

Raises:

DNSError – If the domain name is invalid, or it was impossible to find some IP addresses.

server.app.utils.domain_name_to_ip.edns_response_to_ips(response, client_ip, wanted_ip_type, resolvers, depth=0, max_depth=2)

This method takes the IPs from the response. In case the response has a CNAME, it will recursively try to get an IP from that CNAME. In the worst case, it will be redirected “max_depth” times. This parameter is useful for preventing an infinite loop in redirecting.

Parameters:

• response (dns.message.Message) – The response from the EDNS query.

• client_ip (str) – The client IP.

• wanted_ip_type (int) – The IP type of the resulting IPs that we want.

• resolvers (list) – A list of popular DNS resolvers that are ECS-capable. They are used in the CNAME case.

• depth (int) – The depth of the EDNS query.

• max_depth (int) – The maximum depth of the EDNS query.

Returns:

A list of IPs taken from the response.

Return type:

list(str)

server.app.utils.domain_name_to_ip.perform_edns_query(domain_name, resolver_name, ecs, wanted_ip_type, timeout=3)

This method performs a EDNS query against the domain name using the resolver as the DNS IP and returns the response.

Parameters:

• domain_name (str) – The domain name.

• resolver_name (str) – The resolver name.

• ecs (dns.edns.ECSOption) – The EDNS query option. It contains information about the client IP.

• wanted_ip_type (int) – The IP type of the EDNS query. (4 or 6)

• timeout (float|int) – The timeout for the EDNS query.

Returns:

The response from the EDNS query.

Return type:

Optional[dns.message.Message]

IP utility methods

server.app.utils.ip_utils.client_ip_fetch(request, wanted_ip_type)

Attempts to determine the client’s IP address from the request.

Parameters:

• request (Request) – The FastAPI Request object, containing information about the incoming client request.

• wanted_ip_type (int) – The type of IP address that you want to get (4 or 6).

Returns:

The determined IP address of the client (or a fallback server IP).

Return type:

str | None

Raises:

HTTPException – 503: If neither the client’s IP from headers/request nor the fallback server IP can be successfully resolved.

server.app.utils.ip_utils.get_area_of_ip(ip_country, ip_continent)

This method tries to get the area of an IP address based on its country and continent.

Parameters:

• ip_country (Optional[str]) – The country code of the IP address.

• ip_continent (Optional[str]) – The continent code of the IP address.

Returns:

The area of an IP address.

Return type:

str

server.app.utils.ip_utils.get_ip_family(ip_str)

This method returns the IP family of the given IP address. It returns 4 if we have an IPv4, and it returns 6 if we have an IPv6 address. Otherwise, it raises an exception.

Parameters:

ip_str (Optional[str]) – The IP address

Returns:

The ip family or an exception if we do not get an IP address

Return type:

int

Raises:

InputError – If the IP provided is not an IPv4 or IPv6 address.

server.app.utils.ip_utils.get_ip_network_details(ip_str)

This method gets the ASN, the country code and the continent code of an IP address.

Parameters:

ip_str (str) – The ip address

Returns:

the ASN, the country code and the continent of an IP address if they can be taken.

Return type:

tuple[Optional[str], Optional[str], Optional[str]]

server.app.utils.ip_utils.get_prefix_from_ip(ip_str)

This method returns the prefix of an IP address. It randomizes it before sending it to stat.ripe.net

Parameters:

ip_str (str) – The ip address.

Returns:

the prefix of an IP address.

Return type:

Optional[str]

server.app.utils.ip_utils.get_server_ip(wanted_ip_type)

It determines the public IP address of this server by opening a dummy UDP socket connection to DNS (taken from the config). It has fallbacks to ipify.org. If you want IPv4, it will open an IPv4 connection, otherwise it will open an IPv6 connection. It is strict, and it will return None if it could not return the type you wanted.

Parameters:

wanted_ip_type (int) – The type of IP address we are looking for.

Returns:

The server’s external IP address. as an IPv4Address or IPv6Address object, or None if detection fails.

Return type:

Optional[IPv4Address | IPv6Address]

server.app.utils.ip_utils.get_server_ip_from_ipify(wanted_ip_type)

This method is a fallback to try to get the public IP address of our server from ipify.org

Parameters:

wanted_ip_type (int) – The type of IP address that you want to get. (4 or 6)

Returns:

The public IP address of our server.

Return type:

Optional[IPv4Address | IPv6Address]

server.app.utils.ip_utils.get_server_ip_if_possible(wanted_ip_type)

This method returns the IP address of this server. If it has both IPv6 and IPv4, it will return whatever type you wanted. If not, it returns the type it has. (It has at least one IP address which is either IPv4 or IPv6)

Parameters:

wanted_ip_type (int) – The type of IP address that you want to get. (4 or 6)

Returns:

The IP address of this server with the desired IP type if possible.

Return type:

Optional[IPv4Address | IPv6Address]

server.app.utils.ip_utils.ip_to_str(ip)

Converts an IP address (either IPv4 or IPv6) to its string representation. This function takes an IPv4Address or IPv6Address object and converts it to a string. If the input IP is None, it returns None.

Parameters:

ip (Optional[IPv4Address | IPv6Address]) – The IP address to be converted.

Returns:

The string representation of the IP address, or None if the input is None.

Return type:

Optional[str]

server.app.utils.ip_utils.is_private_ip(ip_str)

This method checks whether an IP address is a private IP.

Parameters:

ip_str (str) – The IP address to check.

Returns:

Whether the IP address is a private IP.

Return type:

bool

server.app.utils.ip_utils.is_this_ip_anycast(searched_ip)

This method checks whether an IP address is anycast or not, by searching in the local anycast prefix databases. This method would never throw an exception (If the databases don’t exist, it will return False).

Parameters:

searched_ip (Optional[str]) – The IP address to check.

Returns:

Whether the IP address is anycast or not.

Return type:

bool

server.app.utils.ip_utils.randomize_ip(ip)

Randomizes the host bits of an IPv4 or IPv6 address based on a subnet mask length.

Parameters:

ip (IPv4Address | IPv6Address) – The IPv4 or IPv6 address to randomize.

Returns:

A new IPv4 or IPv6 address with the same network bits and randomized host bits.

Return type:

IPv4Address | IPv6Address | None

server.app.utils.ip_utils.ref_id_to_ip_or_name(ref_id, stratum, ip_family)

Represents a method that converts the reference id to the reference ip or reference name. If the stratum is 0 or 1 then we can convert the reference id to it’s name (ex: Geostationary Orbit Environment Satellite). If the stratum is between 1 and 256 then we can convert the reference id to it’s IP. But if the ip is an IPV6, then there is an M5 hash involved, and all we can do is to show it (we cannot decode it). If the stratum is greater than 255, then we have an invalid stratum.

Parameters:

• ref_id (int) – the reference id of the ntp server.

• stratum (int) – the stratum level of the ntp server.

• ip_family (int) – the ip family of the ntp server. (4 or 6)

Return type:

tuple[None, str] | tuple[IPv4Address | IPv6Address, None] | tuple[None, None]

Returns:

a tuple of the ip and name of the ntp server. At least one of them is None. If both are None then the stratum is invalid.

server.app.utils.ip_utils.try_converting_ip(client_ip, wanted_ip_type)

This method tries to convert an IPv4 into IPv6 or an IPv6 into an IPv4 using reverse DNS from dnspython. It only works if there is a configured PTR record + AAAA record.

Parameters:

• client_ip (Optional[str]) – The client IP to convert.

• wanted_ip_type (int) – The type of IP address that we want.

Returns:

The converted IPv6 or IPv4 as a string or the original IP if the process failed.

Return type:

Optional[str]

Methods used for fetching data from the config file

server.app.utils.load_config_data.check_geolite_account_id_and_key()

This function checks that we have the account id and key set. Only Warnings. It does not raise errors.

Returns:

True if we have the account id and key set.

Return type:

bool

server.app.utils.load_config_data.get_anycast_prefixes_v4_url()

This method returns the URL prefixes for anycast IPv4 servers.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

str

server.app.utils.load_config_data.get_anycast_prefixes_v6_url()

This method returns the URL prefixes for anycast IPv6 servers.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

str

server.app.utils.load_config_data.get_edns_default_servers()

This method returns the default list of EDNS servers. (in the order of their priorities)

Raises:

ValueError – If this variable has not been correctly set.

Return type:

list[str]

server.app.utils.load_config_data.get_edns_timeout_s()

This method returns the timeout for the EDNS query request.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

float | int

server.app.utils.load_config_data.get_ipv4_edns_server()

This method returns the first IPv4 EDNS server available in the config. It returns None if no IPv4 EDNS server is available.

Return type:

Optional[str]

server.app.utils.load_config_data.get_ipv6_edns_server()

This method returns the first IPv6 EDNS server available in the config. It returns None if no IPv6 EDNS server is available.

Return type:

Optional[str]

server.app.utils.load_config_data.get_mask_ipv4()

This method returns the mask we use for ipv4 IPs.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

int

server.app.utils.load_config_data.get_mask_ipv6()

This method returns the mask we use for ipv6 IPs.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

int

server.app.utils.load_config_data.get_max_mind_path_asn()

This method returns the path to the max_mind ASN database used for geolocation.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

str

server.app.utils.load_config_data.get_max_mind_path_city()

This method returns the path to the max_mind city database used for geolocation.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

str

server.app.utils.load_config_data.get_max_mind_path_country()

This method returns the path to the max_mind country database used for geolocation.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

str

server.app.utils.load_config_data.get_nr_of_measurements_for_jitter()

This method returns the number of measurement requested for calculating the jitter.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

int

server.app.utils.load_config_data.get_ntp_version()

This method returns the ntp version that we use in measurements.

Raises:

ValueError – If the ntp version has not been correctly set.

Return type:

int

server.app.utils.load_config_data.get_rate_limit_per_client_ip()

This method returns the rate limit for queries per client IP to our server.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

str

server.app.utils.load_config_data.get_ripe_account_email()

This function returns the RIPE Atlas account email. (one that has enough credits)

Raises:

ValueError – If the RIPE Atlas account email is not set.

Return type:

str

server.app.utils.load_config_data.get_ripe_api_token()

This function returns the RIPE Atlas API token.

Raises:

ValueError – If the RIPE Atlas API token is not set.

Return type:

str

server.app.utils.load_config_data.get_ripe_number_of_probes_per_measurement()

This method returns the number of probes requested and desired for a measurement.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

int

server.app.utils.load_config_data.get_ripe_packets_per_probe()

This method returns the number of tries that a probe will do for a measurement. It will send “packets_per_probe” queries for that NTP server. (see RIPE Atlas documentation for more information)

Raises:

ValueError – If this variable has not been correctly set.

Return type:

int

server.app.utils.load_config_data.get_ripe_server_timeout()

This method returns the timeout (seconds) that the server has to get data from RIPE Atlas.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

int

server.app.utils.load_config_data.get_ripe_timeout_per_probe_ms()

This method returns the timeout that a probe has to receive an answer from a measurement.

Raises:

ValueError – If this variable has not been correctly set.

Return type:

float | int

server.app.utils.load_config_data.get_timeout_measurement_s()

This method returns the timeout for an NTP measurement.

Raises:

ValueError – If the ntp version has not been correctly set.

Return type:

float | int

server.app.utils.load_config_data.load_config()

It loads the config from a YAML file.

Raises:

FileNotFoundError – If the config file does not exist.

Return type:

dict[str, Any]

server.app.utils.load_config_data.verify_if_config_is_set()

This method ensures that the config file has all the required variables, and they are all correctly set. (the same data type) It will return true if everything is fine, else it will rise an exception,

Raises:

ValueError – If the config file does not have all the required variables or some of them are invalid.

Return type:

bool

Methods used fetching geoloaction information from MaxMind databases

server.app.utils.location_resolver.get_asn_for_ip(client_ip)

Retrieves the asn of a given IP address using the MaxMind GeoLite2-ASN database. If the asn cannot be determined due to missing data, database issues, or the IP not being found, it returns None.

Parameters:

client_ip (str) – The IP address of the client to geolocate.

Returns:

The ans for the ip location or None.

Return type:

Optional[str]

server.app.utils.location_resolver.get_continent_for_ip(client_ip)

Retrieves the continent code of a given IP address using the MaxMind GeoLite2-Country database. If the continent code cannot be determined due to missing data, database issues, or the IP not being found, it returns None.

Parameters:

client_ip (Optional[str]) – The IP address of the client to geolocate.

Returns:

The continent code for the ip location or None.

Return type:

Optional[str]

server.app.utils.location_resolver.get_coordinates_for_ip(client_ip)

Retrieves the geographical location (latitude and longitude) of a given IP address using the MaxMind GeoLite2-City database.

If the location cannot be determined due to missing data, database issues, or the IP not being found, it returns the coordinates of an (almost) randomly chosen location as a fallback.

Parameters:

client_ip (Optional[str]) – The IP address of the client to geolocate

Returns:

A tuple containing the latitude and longitude. If the location cannot be resolved, returns (25.0, -71.0).

Return type:

tuple[float, float]

server.app.utils.location_resolver.get_country_for_ip(client_ip)

Retrieves the country code of a given IP address using the MaxMind GeoLite2-Country database. If the country code cannot be determined due to missing data, database issues, or the IP not being found, it returns None.

Parameters:

client_ip (Optional[str]) – The IP address of the client to geolocate.

Returns:

The country code for the ip location or None.

Return type:

Optional[str]

Business logic for the measurement engine

server.app.utils.perform_measurements.convert_ntp_response_to_measurement(response, server_ip_str, server_name, ntp_version=4)

This method converts an NTP response to an NTP measurement object.

Parameters:

• response (ntplib.NTPStats) – The NTP response to convert.

• server_ip_str (str) – The IP address of the ntp server in string format.

• server_name (Optional[str]) – The name of the ntp server.

• ntp_version (int) – The version of the ntp that you want to use.

Returns:

It returns an NTP measurement object if the conversion was successful.

Return type:

Optional[NtpMeasurement]

server.app.utils.perform_measurements.get_request_settings(ip_family_of_ntp_server, ntp_server, client_ip, probes_requested=3)

This method gets the RIPE measurement settings for the performing a RIPE measurement. :type ip_family_of_ntp_server: int :param ip_family_of_ntp_server: The IP family of the NTP server. (4 or 6) :type ip_family_of_ntp_server: int :type ntp_server: str :param ntp_server: The NTP server IP address or domain name. :type ntp_server: str :type client_ip: str :param client_ip: The IP address of the client. :type client_ip: str :type probes_requested: int :param probes_requested: The number of probes requested. :type probes_requested: int

Returns:

Returns the RIPE measurement settings for the performing a RIPE measurement.

Return type:

tuple[dict, dict]

Raises:

• InputError – If the input is invalid.

• RipeMeasurementError – If the ripe measurement could not be performed.

• ValueError – If some variable in env is not correctly set.

server.app.utils.perform_measurements.perform_ntp_measurement_domain_name_list(server_name, client_ip=None, wanted_ip_type=4, ntp_version=4)

This method performs a NTP measurement on a NTP server from all the IPs got back from its domain name.

Parameters:

• server_name (str) – The name of the ntp server.

• client_ip (Optional[str]) – The IP address of the client (if given).

• wanted_ip_type (int) – The IP type that we want to measure.

• ntp_version (int) – The version of the ntp that you want to use.

Returns:

It returns a list of NTP measurement objects or None if there is a timeout.

Return type:

Optional[list[NtpMeasurement]]

Raises:

DNSError – If the domain name is invalid or cannot be converted to an IP list.

server.app.utils.perform_measurements.perform_ntp_measurement_ip(server_ip_str, ntp_version=4)

This method performs an NTP measurement on an NTP server using its IP address.

Parameters:

• server_ip_str (str) – The IP address of the ntp server in string format.

• ntp_version (int) – The version of the ntp that you want to use.

Returns:

It returns the NTP measurement object or None if something wrong happened. (usually timeouts)

Return type:

Optional[NtpMeasurement]

server.app.utils.perform_measurements.perform_ripe_measurement_domain_name(server_name, client_ip, wanted_ip_type, probes_requested=3)

This method performs a RIPE measurement on a domain name. It lets the RIPE atlas probe to decide which IP of that domain name to use. (You can see this IP from the details of the measurement by looking at the measurement ID)

Parameters:

• server_name (str) – The domain name of the NTP server.

• client_ip (str) – The IP address of the NTP server.

• wanted_ip_type (int) – The IP type that we want to measure.

• probes_requested (int) – The number of probes requested.

Returns:

It returns the ID of the measurement and the list of IPs of the domain name.

You can find in the measurement what IP it used.

Return type:

int

Raises:

• InputError – If the conversion could not be performed.

• RipeMeasurementError – If the ripe measurement could not be performed.

server.app.utils.perform_measurements.perform_ripe_measurement_ip(ntp_server_ip, client_ip, probes_requested=3)

This method performs a RIPE measurement and returns the ID of the measurement.

Parameters:

• ntp_server_ip (str) – The NTP server IP.

• client_ip (str) – The IP of the client.

• probes_requested (int) – The number of probes requested.

Returns:

The ID of the measurement.

Return type:

int

Raises:

• InputError – If the NTP server IP is not valid, probe requested is negative.

• RipeMeasurementError – If the ripe measurement could not be performed.

server.app.utils.perform_measurements.print_ntp_measurement(measurement)

It prints the ntp measurement in a human-readable format and returns True if the printing was successful.

Parameters:

measurement (NtpMeasurement) – The NtpMeasurement object.

Return type:

bool

Methods used for fetching and parsing data from RIPE Atlas

server.app.utils.ripe_fetch_data.check_all_measurements_done(measurement_id, measurement_req)

Check the status of a RIPE Atlas measurement.

This function queries the RIPE Atlas API for a given measurement ID and determines whether the measurement is complete, ongoing, or should be considered timed out. The status is based on the number of probes requested, the measurement status provided by RIPE, and how much time has passed since the measurement started.

Parameters:

• measurement_id (str) – RIPE Atlas measurement ID.

• measurement_req (int) – The number of probes expected for the measurement to be considered complete.

Returns:

• “Complete”: All expected probes have responded or the measurement is stopped.

• ”Ongoing”: The measurement is still in progress and has not exceeded the timeout threshold.

• ”Timeout”: The measurement did not complete within the allowed time window.

Return type:

str

Raises:

RipeMeasurementError – If there are errors with the response from ripe, either not received or malformed.

Notes

• If the difference between the current time and the measurement’s start time exceeds the configured time in seconds, and the measurement is not yet complete, it is considered “Timeout”.

• This function assumes a successful HTTP response from the RIPE API; if not, it will raise an exception.

server.app.utils.ripe_fetch_data.check_all_measurements_scheduled(measurement_id)

Check if all RIPE Atlas probes for a given measurement have been scheduled.

This function compares the number of probes requested with the number of probes successfully scheduled. If all requested probes are scheduled, it returns True. Otherwise, it returns False

Parameters:

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

Returns:

True if all requested probes have been scheduled, False otherwise.

Return type:

bool

Raises:

ValueError – If the RIPE API returns an error or if the probe count data is missing or invalid.

server.app.utils.ripe_fetch_data.get_data_from_ripe_measurement(measurement_id)

Fetches raw measurement results from the RIPE Atlas API.

This function queries the RIPE Atlas v2 API using the provided measurement ID, authenticates with an API key stored in the RIPE_KEY environment variable, and returns the parsed JSON response as a list of dictionaries.

Parameters:

measurement_id (str) – The RIPE Atlas measurement ID to fetch results for.

Returns:

A list of measurement result entries as dictionaries.

Return type:

list[dict[str, Any]]

Raises:

RipeMeasurementError – If the HTTP request fails or the response cannot be parsed as JSON or the answer is not a list of dicts.

Notes

• Requires the RIPE_KEY environment variable to be set with a valid API key.

server.app.utils.ripe_fetch_data.get_probe_data_from_ripe_by_id(probe_id)

Retrieves detailed information about a specific RIPE Atlas probe by its ID.

This function sends a GET request to the RIPE Atlas API to fetch metadata about the specified probe. The request is authenticated using the RIPE API key stored in the RIPE_KEY environment variable.

Parameters:

probe_id (str) – The ID of the RIPE Atlas probe to fetch information for.

Returns:

A dictionary containing metadata about the probe

Return type:

dict[str, Any]

Raises:

RipeMeasurementError – If the HTTP request fails or the response is not valid JSON.

Notes

• Requires the RIPE_KEY environment variable to be set with a valid API key.

server.app.utils.ripe_fetch_data.is_failed_measurement(entry)

Determines if a RIPE measurement entry has failed.

A measurement is considered failed if all result entries contain the key “x” with value “*”, which typically indicates a failed probe response.

Parameters:

entry (dict[str, Any]) – A dictionary representing a single measurement entry from the RIPE API.

Returns:

True if all entries in the result indicate failure, False otherwise.

Return type:

bool

server.app.utils.ripe_fetch_data.parse_data_from_ripe_measurement(data_measurement)

Parses raw RIPE Atlas measurement data into a list of RipeMeasurement objects.

This function:

• Determines whether each measurement entry failed or succeeded.

• Extracts NTP-related server and timing information.

• Converts timestamps and metrics into structured internal representations.

• Adds probe-specific metadata fetched from the RIPE API.

• Returns a status string indicating whether all measurements have been processed.

Parameters:

data_measurement (list[dict]) – A list of dictionaries representing raw measurement entries from the RIPE Atlas API.

Returns:

• A list of parsed and structured RipeMeasurement objects.

• A status string returned by check_all_measurements_done, indicating whether all expected measurements are present and complete.

Return type:

tuple[list[RipeMeasurement], str]

Notes

• Measurements that are marked as failed are still processed, but filled with default values.

• Probe metadata is fetched using the probe ID (prb_id) in each measurement.

• Timestamps are converted using convert_float_to_precise_time.

server.app.utils.ripe_fetch_data.parse_probe_data(probe_response)

Parses probe metadata received from the RIPE Atlas API into a ProbeData object.

This function extracts relevant information such as the probe ID, IP addresses, country code, and coordinates from the API response.

Parameters:

probe_response (dict) – The raw dictionary response from the RIPE Atlas probe lookup API.

Returns:

Parsed information including the probe’s ID, IP addresses, country code, and location.

Return type:

ProbeData

Notes

• If an error is present in the response, a default ProbeData with dummy values is returned.

• Coordinates default to [0.0, 0.0] if not provided.

• Country code defaults to “NO COUNTRY CODE” if not found.

server.app.utils.ripe_fetch_data.successful_measurement(entry)

Identifies the index of the successful measurement with the lowest offset.

A successful measurement is one that contains the “origin-ts” field, indicating a valid timestamp. Among all such entries, the one with the lowest absolute offset value is considered the most successful.

Parameters:

entry (dict[str, Any]) – A dictionary representing a single measurement entry from the RIPE API.

Returns:

The index of the result entry with the lowest absolute offset (closest to 0), or None if no valid entries exist.

Return type:

int | None

Methods used for selecting the probes used for RIPE Atlas

server.app.utils.ripe_probes.consume_probes(probes_requested, current_probes_set, probes_ids)

This method takes how many probes we need from probes_ids and put them in current_probes_set. If we found enough probes, we ignore the remaining probes.

Parameters:

• probes_requested (int) – The number of probes that we still need to request before calling this method.

• current_probes_set (set[int]) – The set of probes we are currently requesting.

• probes_ids (list[int]) – A list with the ids of the probes that we found available, and we want to add them.

Returns:

• The remained number of probes still to find.

• The updated set of probes that we will use in the measurement.

Return type:

tuple[int, set[int]]

server.app.utils.ripe_probes.get_area_probes(area, n)

This method selects n probes from this area.

Parameters:

• area (Optional[str]) – The area of the probes.

• n (int) – The number of probes to select.

Returns:

The selected probes.

Return type:

dict

Raises:

InputError – If area is not valid.

server.app.utils.ripe_probes.get_asn_probes(ip_asn, n)

This method selects n probes that belong to the same ASN network.

Parameters:

• ip_asn (Optional[str | int]) – The ASN network.

• n (int) – The number of probes to select.

Returns:

The selected probes.

Return type:

dict

Raises:

InputError – If the ASN network is None.

server.app.utils.ripe_probes.get_available_probes_asn(client_ip, ip_asn, ip_type)

This method gets the probes available on RIPE Atlas that has the same ASN as the client IP. These probes should also support ipv4 or ipv6, it depends on the type.

Parameters:

• client_ip (str) – The IP address of the client.

• ip_asn (str) – The ASN of the searched network.

• ip_type (str) – The IP type (ipv4 or ipv6). (not case-sensitive)

Returns:

A list with the ids of the available probes.

Return type:

list[int]

Raises:

Exception – If the input is invalid.

server.app.utils.ripe_probes.get_available_probes_asn_and_country(client_ip, ip_asn, ip_country_code, ip_type)

This method gets the probes available on RIPE Atlas that has the same ASN and country as the client IP. These probes should also support ipv4 or ipv6, it depends on the type.

Parameters:

• client_ip (str) – The IP address of the client.

• ip_asn (str) – The ASN of the searched network.

• ip_country_code (str) – The country code of the respective IP.

• ip_type (str) – The IP type (ipv4 or ipv6). (not case-sensitive)

Returns:

A list with the ids of the available probes.

Return type:

list[int]

Raises:

Exception – If the input is invalid.

server.app.utils.ripe_probes.get_available_probes_asn_and_prefix(client_ip, ip_asn, ip_prefix, ip_type)

This method gets the probes available on RIPE Atlas that has the same ASN and prefix as the client IP. These probes should also support ipv4 or ipv6, it depends on the type.

Parameters:

• client_ip (str) – The IP address of the client.

• ip_asn (str) – The ASN of the searched network.

• ip_prefix (str) – The prefix of the respective IP.

• ip_type (str) – The IP type (ipv4 or ipv6). (not case-sensitive)

Returns:

A list with the ids of the available probes.

Return type:

list[int]

Raises:

Exception – If the input is invalid.

server.app.utils.ripe_probes.get_available_probes_country(client_ip, country_code, ip_type)

This method gets the probes available on RIPE Atlas that has the same country as the client IP. These probes should also support ipv4 or ipv6, it depends on the type.

Parameters:

• client_ip (str) – The IP address of the client.

• country_code (str) – The country code.

• ip_type (str) – The IP type (ipv4 or ipv6). It should be lowercase.

Returns:

A list with the ids of the available probes.

Return type:

list[int]

Raises:

Exception – If the input is invalid.

server.app.utils.ripe_probes.get_available_probes_prefix(client_ip, ip_prefix, ip_type)

This method gets the probes available on RIPE Atlas that has the same prefix as the client IP. These probes should also support ipv4 or ipv6, it depends on the type.

Parameters:

• client_ip (str) – The IP address of the client.

• ip_prefix (str) – The ip_prefix of the searched network.

• ip_type (str) – The IP type (ipv4 or ipv6). It should be lowercase.

Returns:

A list with the ids of the available probes.

Return type:

list[int]

Raises:

Exception – If the input is invalid.

server.app.utils.ripe_probes.get_best_probes_matched_by_single_attribute(client_ip, current_probes_set, ip_asn, ip_prefix, ip_country, ip_family, probes_requested=3)

This method is responsible for getting the best probes that has a match by a single attribute in this order: ASN, prefix, country. As soon as we have enough probes we return.

Parameters:

• client_ip (str) – The IP address of the client.

• current_probes_set (set[int]) – The set of probes that we will use for the measurement. (to be sure that we do not include duplicates)

• ip_asn (Optional[str]) – The ASN of the NTP server IP address.

• ip_prefix (Optional[str]) – The prefix of the NTP server IP address.

• ip_country (Optional[str]) – The country of the NTP server IP address.

• ip_family (int) – The family of the NTP server IP address. (4 or 6)

• probes_requested (int) – The number of probes that we still need to request.

Returns:

• The updated number of probes that we still need to find after this method call.

  • The set of probe types and the respective number of probes.

Return type:

tuple[int, set[int]]

Raises:

InputError – If the NTP server IP is invalid or if the probes_requested is negative.

server.app.utils.ripe_probes.get_best_probes_with_multiple_attributes(client_ip, current_probes_set, ip_asn, ip_prefix, ip_country, ip_family, probes_requested=3)

This method tries to get probes that has the same ASN and prefix OR the same ASN and country and subtract them from the probes_requested. These probes have the highest priority as they have multiple attributes as the client IP (same ASN and same prefix OR same ASN or same country)

Parameters:

• client_ip (str) – The IP address of the client.

• current_probes_set (set[int]) – The set of probes that we will use for the measurement. (to be sure that we do not include duplicates)

• ip_asn (Optional[str]) – The ASN of the NTP server IP address.

• ip_prefix (Optional[str]) – The prefix of the NTP server IP address.

• ip_country (Optional[str]) – The country of the NTP server IP address.

• ip_family (int) – The family of the NTP server IP address. (4 or 6)

• probes_requested (int) – The number of probes that we still need to request.

Returns:

The updated number of probes that we still need to find after this method call.

a list of the IDs of the probes that we found until now.

Return type:

tuple[int, set[int]]

Raises:

InputError – If the input is invalid or probes_requested is negative.

server.app.utils.ripe_probes.get_country_probes(ip_country_code, n)

This method selects n probes that belong to the same country.

Parameters:

• ip_country_code (Optional[str]) – The country code.

• n (int) – The number of probes to select.

Returns:

The selected probes.

Return type:

dict

Raises:

InputError – If the country code is None.

server.app.utils.ripe_probes.get_prefix_probes(ip_prefix, n)

This method selects n probes that has this prefix.

Parameters:

• ip_prefix (Optional[str]) – The IP prefix family.

• n (int) – The number of probes to select.

Returns:

The selected probes.

Return type:

dict

Raises:

InputError – If the IP prefix is None.

server.app.utils.ripe_probes.get_probes(client_ip, ip_family_of_ntp_server, probes_requested=3)

This method handles all cases regarding what probes we should send. This method assumes all inputs are either valid or None. (If there is a typo in the input, the measurement may be affected) It will try to return the best probes near the client.

Parameters:

• client_ip (str) – The IP address of the client.

• ip_family_of_ntp_server (int) – The IP family of the NTP server. (4 or 6)

• probes_requested (int) – The total number of probes that we will request.

Returns:

The list of probes that we will use for the measurement.

Return type:

list[dict]

Raises:

InputError – If the client IP address is invalid.

server.app.utils.ripe_probes.get_probes_by_ids(probe_ids)

This method selects probes by their IDs.

Parameters:

probe_ids (list[int]) – The IDs of the probes.

Returns:

The selected probes.

Return type:

dict

Raises:

InputError – If the input is invalid.

server.app.utils.ripe_probes.get_random_probes(n)

This method selects n random probes from all over the world.

Parameters:

n (int) – The number of probes to select.

Returns:

The selected probes.

Return type:

dict

Methods used for input validation

server.app.utils.validate.ensure_utc(dt)

Ensures that a given datetime object is timezone-aware and in UTC.

This function checks if the provided datetime is naive (i.e., has no timezone info). If so, it assigns UTC as the timezone. If it already has a timezone, it converts it to UTC.

Parameters:

dt (datetime) – A Python datetime object, either naive or timezone-aware.

Returns:

A timezone-aware datetime object in UTC.

Return type:

datetime

server.app.utils.validate.is_ip_address(ip_str)

It verifies if the given string is a valid IPv4 or IPv6 address. If not, it returns None.

Parameters:

ip_str (Optional[str]) – ip address

Return type:

str | None

Returns:

str | None

server.app.utils.validate.is_valid_domain_name(domain_name)

This method verifies if the given string could be a valid domain name. It verifies the syntax. It offers only a basic validation of the syntax. It is just enough for a dns to consider it.

Parameters:

domain_name (str) – The string to check.

Returns:

True if the string could be a valid domain name, False otherwise.

Return type:

bool

server.app.utils.validate.parse_ip(ip_str)

Parses and validates a string as an IPv4 or IPv6 address.

Attempts to interpret the provided string as a valid IP address using ipaddress.ip_address().

Parameters:

ip_str (str) – The IP address in string format (e.g., “192.168.0.1” or “::1”).

Returns:

• A valid IPv4Address or IPv6Address object if parsing succeeds.

• None if the string is not a valid IP address.

Return type:

ipaddress.IPv4Address | ipaddress.IPv6Address | None

server.app.utils.validate.sanitize_string(s)

This method sanitize a string. It removes null characters () or control characters (
-) or the DEL character () from a string. It removes them. We do this because our database does not allow these characters (security concerns). It prints a warning if the string contains such characters.

Parameters:

s (Optional[str]) – The string to sanitize.

Returns:

The sanitized string, or None if the string is None.

Return type:

Optional[str]