Onionprobe

https://onionservices.torproject.org/apps/web/onionprobe/
Admin
Onionprobe is a tool for testing and monitoring the status of Tor Onion Services sites, capable of...
Tokens:54,131
Snippets:281
Trust Score:10
Update:3 months ago
Show doc for...
Tokens:
Raw
### Install Python dependencies for source

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Sets up a virtual environment and installs dependencies from PyPI.

```bash
sudo apt install -y python3-pip tor
python3 -m venv venv
source venv/bin/activate
pip3 install .
```

--------------------------------

### Install Onionprobe from source

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Clones the repository and installs the package using pip.

```bash
sudo apt install -y python3-pip
git clone https://gitlab.torproject.org/tpo/onion-services/onionbalance
cd onionprobe
python3 -m pip install . --break-system-packages
```

--------------------------------

### Install Onionprobe from source system-wide

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Installs the source package system-wide using sudo.

```bash
sudo python3 -m pip install . --break-system-packages
```

--------------------------------

### Install Onionprobe via pipx

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

The recommended method for installing the Python package.

```bash
pipx install onionprobe
```

--------------------------------

### Install Onionprobe via virtualenv

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Sets up a virtual environment and installs Onionprobe within it.

```bash
mkdir onionprobe
cd onionprobe
python3 -m venv venv
source venv/bin/activate
pip install onionprobe
```

--------------------------------

### Install Onionprobe on Arch Linux

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Installs the Onionprobe package from the AUR.

```bash
pacman -S onionprobe-git
```

--------------------------------

### Start and Monitor the Node

Source: https://onionservices.torproject.org/apps/web/onionprobe/standalone

Commands to start the monitoring node in the background and view container logs.

```bash
./onionprobe-monitor up   # This will fork into the background after bootstrap
./onionprobe-monitor logs # View container logs
```

--------------------------------

### Upload and verify packages

Source: https://onionservices.torproject.org/apps/web/onionprobe/development

Upload to PyPI and verify installation from repositories.

```bash
make upload-python-test-package

```

```bash
sudo apt-get install -y python3-pip tor
pip install -i https://pypi.org/simple/ \
            --extra-index-url https://test.pypi.org/simple \
            --break-system-packages \
            onionprobe==$ONIONPROBE_VERSION

```

```bash
make upload-python-package

```

```bash
sudo apt-get install -y python3-pip tor
pip install --break-system-packages \
  onionprobe==$ONIONPROBE_VERSION

```

--------------------------------

### Build and test Python package

Source: https://onionservices.torproject.org/apps/web/onionprobe/development

Build the package and verify installation in a fresh environment.

```bash
make build-python-package

```

```bash
sudo apt-get install -y python3-pip tor
pip install --break-system-packages \
  dist/onionprobe-$ONIONPROBE_VERSION-*.whl

```

```bash
$HOME/.local/bin/onionprobe --version
$HOME/.local/bin/onionprobe -e \
  http://2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion
$HOME/.local/bin/onionprobe -e \
  https://v236xhqtyullodhf26szyjepvkbv6iitrhjgrqj4avaoukebkk6n6syd.onion

```

--------------------------------

### Install Debian dependencies for source

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Installs required system dependencies for running from source on Debian.

```bash
sudo apt install -y python3-prometheus-client python3-stem \
                    python3-cryptography python3-yaml      \
                    python3-requests python3-socks tor
```

--------------------------------

### initialize_prometheus_exporter()

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Initializes and starts the Prometheus HTTP exporter server.

```APIDOC
## initialize_prometheus_exporter()

### Description
Starts the Prometheus HTTP exporter server on the port configured in the application settings.
```

--------------------------------

### Initialize the Prometheus Exporter

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Starts the Prometheus HTTP server on the configured port.

```python
def initialize_prometheus_exporter(self):
    """
    Initialize the Prometheus Exporter
    """

    from prometheus_client import start_http_server

    port = self.get_config('prometheus_exporter_port')

    self.log('Initializing Prometheus HTTP exporter server at port %s...' % (port))
    start_http_server(port)
```

--------------------------------

### Stop and Start Monitoring Node

Source: https://onionservices.torproject.org/apps/web/onionprobe/upgrading

Commands to stop and start the standalone monitoring node during maintenance.

```bash
./onionprobe-monitor down
```

```bash
./onionprobe-monitor start
```

--------------------------------

### Provision Onionprobe script

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Runs the convenience script to install dependencies and Tor.

```bash
./scripts/provision-onionprobe
```

--------------------------------

### Configure Onionprobe MetricsPort

Source: https://onionservices.torproject.org/apps/web/onionprobe/standalone

Examples for setting metrics_port and metrics_port_policy in the Onionprobe configuration file.

```yaml
# The following should work by default for containers in the
# 172.16.0.0/12 subnet.
metrics_port: '0.0.0.0:9936'
metrics_port_policy: 'accept 172.16.0.0/12'
```

```yaml
# The following should work by default for a local network, including local
# containers (not recommended):
metrics_port: '0.0.0.0:9936'
metrics_port_policy: 'accept 192.168.0.0/16,accept 10.0.0.0/8,accept 172.16.0.0/12'
```

```yaml
# This will allow only the host 172.19.0.100 to connect, and requires
# that the Prometheus service containers binds to this IP address.
metrics_port: '172.19.0.100:9936'
metrics_port_policy: 'accept 172.19.0.100'
```

--------------------------------

### Install Onionprobe on Debian

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Installs Onionprobe and required dependencies including the Tor daemon.

```bash
sudo apt install onionprobe
```

--------------------------------

### Install Onionprobe dependencies on Debian

Source: https://onionservices.torproject.org/apps/web/onionprobe/requirements

Use this command to install all required software packages on a Debian-based system.

```bash
sudo apt install python3 python3-prometheus-client \
                 python3-stem python3-cryptography \
                 python3-yaml python3-requests     \
                 python3-socks tor
```

--------------------------------

### Setup New Tor Circuit

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Closes the existing circuit and establishes a new one to attach to a specific stream.

```python
def new_circuit(self, stream):
        """
        Setup a fresh Tor circuit for new streams

        See https://stem.torproject.org/tutorials/to_russia_with_love.html

        :type  stream: stem.response.events.StreamEvent
        :param stream: Stream event
        """

        self.log('Building new circuit...', 'debug')

        # Remove the old circuit
        if self.circuit_id is not None:
            self.log('Removing old circuit {}...'.format(self.circuit_id), 'debug')
            self.controller.close_circuit(self.circuit_id)

        # Create new circuit
        self.circuit_id = self.controller.new_circuit(await_build=True)

        # Attach the new stream
        if stream.status == 'NEW':
            self.log('Setting up new circuit {}...'.format(self.circuit_id), 'debug')
            self.controller.attach_stream(stream.id, self.circuit_id)
```

--------------------------------

### Run with a configuration file

Source: https://onionservices.torproject.org/apps/web/onionprobe/usage

Use the -c flag to specify a YAML configuration file containing endpoints and options.

```bash
onionprobe -c configs/tor.yaml
```

--------------------------------

### Install Tor daemon on Debian

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Required for Python package installations as they do not include the Tor daemon.

```bash
sudo apt install -y tor
```

--------------------------------

### Initialize the metrics subsystem

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Sets up the internal metrics object and initializes default state metrics.

```python
def initialize_metrics(self):
    """
    Initialize the metrics subsystem

    It uses Prometheus metrics even if the Prometheus exporter is not in use.

    This means that the Prometheus metrics are always used, even if only for
    internal purposes, saving resources from preventing us to build additional
    metric logic.
    """

    # The metrics object
    self.metrics = metrics

    # Set version
    self.metrics['onionprobe_version'].info({
        'version': onionprobe_version,
        })

    # Set initial state
    self.metrics['onionprobe_state'].state('starting')
```

--------------------------------

### Install Onionprobe via pip without virtualenv

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Direct installation which may conflict with system-wide packages.

```bash
pip install onionprobe --break-system-packages
```

--------------------------------

### run(args)

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Initializes and executes the Onionprobe instance using the provided configuration arguments.

```APIDOC
## run(args)

### Description
Run Onionprobe from arguments.

### Parameters
- **args** (dict) - Required - Instance arguments.
```

--------------------------------

### Run Onionprobe as an installed package

Source: https://onionservices.torproject.org/apps/web/onionprobe/api

Use this method when Onionprobe is installed in the Python environment. It imports the command line runner directly from the package.

```python
from onionprobe.app import run_from_cmdline

if __name__ == "__main__":
    run_from_cmdline()
```

--------------------------------

### Main compilation procedure

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Entry point for the main compilation procedure.

```python
def compile(self):
        """
        Main compilation procedure.

        """
```

--------------------------------

### Launch Tor with Configuration

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Configures and launches a Tor process using stem.process.launch_tor_with_config.

```python
# Log config
            #config['Log'] = [
            #    'DEBUG  stdout',
            #    'INFO   stdout',
            #    'NOTICE stdout',
            #    'WARN   stdout',
            #    'ERR    stdout',
            #    ]

            if metrics_port is not None and metrics_port != '' and metrics_port != 0:
                config['MetricsPort'] = str(metrics_port)

            if metrics_port_policy is not None and metrics_port_policy != '':
                config['MetricsPortPolicy'] = str(metrics_port_policy)

            self.tor = stem.process.launch_tor_with_config(
                    config           = config,
                    init_msg_handler = print_bootstrap_lines,
                    )

        except OSError as e:
            self.log(e, 'error')

            return False
```

--------------------------------

### OnionprobeConfigCompiler.__init__

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Constructor for the OnionprobeConfigCompiler class. Initializes the configuration template, output directory, and operational parameters.

```APIDOC
## __init__(databases, config_template, output_folder, wait=0, config_overrides=None, loop=False)

### Description
Initializes the OnionprobeConfigCompiler with the provided data sources, template path, and output directory.

### Parameters
- **databases** (dict) - Required - Dictionary of data sources to fetch .onion sites. Format: { 'database_name': 'database_url' }
- **config_template** (str) - Required - Configuration file path to be used as template
- **output_folder** (str) - Required - Output folder where configs are written
- **wait** (int) - Optional - Time to wait between builds
- **config_overrides** (list) - Optional - List of configuration overrides
- **loop** (bool) - Optional - Whether to run in a continuous loop
```

--------------------------------

### Get current timestamp

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Returns the current system time as a POSIX timestamp.

```python
def timestamp(self):
    """
    Wrapper around datetime.now().timestamp()

    :rtype: datetime.datetime
    :return: Current time.
    """

    return datetime.now().timestamp()
```

--------------------------------

### Initialize OnionprobeConfigCompiler

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Constructor for the OnionprobeConfigCompiler class. Loads the default configuration template and initializes database and output folder settings.

```python
def __init__(self, databases, config_template, output_folder, wait=0, config_overrides=None, loop=False):
    """
    Constructor for the OnionprobeConfigCompiler class.

    Loads the default Onionprobe configuration to be used as a template.

    Keeps the dictionary of Onion Services databases as a class attribute.

    :type  databases: dict
    :param databases: Dictionary of data sources to fetch .onion sites.
                      Format is { 'database_name': 'database_url' }

    :type  config_template: str
    :param config_template: Configuration file path to be used as template

    :type  output_folder: str
    :param output_folder: Output folder where configs are written
    """

    # Initialize the configuration object
    self.config = {}

    # Save the databases of Onion Services
    self.databases = databases

    # Load the default configuration file as a template
    if os.path.exists(config_template):
        print('Loading configuration template from %s...' % (config_template))

        with open(config_template, 'r') as base_config:
            self.config = yaml.load(base_config, yaml.CLoader)

    else:
        raise FileNotFoundError(config_template)

    # Set the output folder
    self.output_folder = output_folder

    if not os.path.exists(output_folder):
        raise FileNotFoundError(output_folder)

    # Set wait time
    self.wait = wait

    # Set loop configuration
    self.loop = loop

    # Apply overrides
    if config_overrides != None:
        # Copy item by item, ensuring type casting
        for item in config_overrides:
            override = item.split('=')

            if len(override) != 2:
                print('Skipping malformed override param %s...' % (item))
                continue

            key   = str(override[0])
            value = override[1]

            if key not in config:
                print('Skipping unknown parameter %s...' % (key))
                continue

            cast = type(config[key]['default'])

            if cast == bool:
                value.lower()

                self.config[key] = True if value == 'true' else False
            else:
                self.config[key] = cast(value)

            print('Setting %s to %s.' % (key, value))
```

--------------------------------

### Get current time

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Returns the current system time as a datetime object.

```python
def now(self):
    """
    Wrapper around datetime.now()

    :rtype: datetime.datetime
    :return: Current time.
    """

    return datetime.now()
```

--------------------------------

### Regenerate manpages

Source: https://onionservices.torproject.org/apps/web/onionprobe/development

Build and verify manual pages.

```bash
make manpages

```

```bash
man -l docs/man/onionprobe.1

```

--------------------------------

### RealWorldOnionSites class implementation

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/helpers

Handles the 'Real-World Onion Sites' database by inheriting from OnionprobeConfigCompiler.

```python
class RealWorldOnionSites(OnionprobeConfigCompiler):
    """
    Handles the 'Real-World Onion Sites' database

    Inherits from the OnionprobeConfigCompiler class, implementing
    custom procedures.
    """

    def build_endpoints_config(self, database):
        """
        Overrides OnionprobeConfigCompiler.build_endpoints_config()
        method with custom logic.

        :type database : str
        :param database: A database name from the databases dictionary.

        :rtype: dict
        :return: Onion Service endpoints in the format accepted by Onionprobe.

        """

        endpoints = {}

        # Get the Onion Service database from a remote CSV file
        if os.path.exists(self.databases[database]):
            print('Using list of %s database endpoints from %s...' % (
                database, self.databases[database]))

            with open(self.databases[database], 'r') as result:
                data = csv.DictReader(result.readlines())

        else:
            try:
                print('Fetching remote list of %s database endpoints from %s...' % (database, self.databases[database]))

                result = requests.get(self.databases[database])
                data   = csv.DictReader(StringIO(result.text))

            except Exception as e:
                # Log the exception
                print(repr(e))

                # Some error happened: do not proceed generating the config
                exit(1)

        # Parse the database and convert it to the Onionprobe endpoints format
        for item in data:
            print('Processing %s...' % (item['site_name']))

            url      = urllib.parse.urlparse(item['onion_url'])
            address  = url.netloc
            protocol = url.scheme if url.scheme != '' else 'http'
            port     = 80 if protocol == 'http' else 443
            paths    = [{
                'path'            : url.path if url.path != '' else '/',
                'allowed_statuses': [ 200 ],
                }]

            # Append to the endpoints dictionary
            if item['site_name'] not in endpoints:
                endpoints[item['site_name']] = {
                        'address' : address,
                        'protocol': protocol,
                        'port'    : port,
                        'paths'   : paths,
                        }

        return endpoints
```

--------------------------------

### Get Certificate Information

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Retrieves basic X.509 certificate details in either tree or flat format.

```python
def get_cert_info(self, cert, format = 'tree'):
    """
    Get basic information from a X.509 certificate.

    This method is compatible with SSLSocket.geetpeercert, with the
    advantage that it does not require a valid certificate in order
    to process it's data.

    :type  cert: cryptography.x509.Certificate
    :param cert: The X.509 Certificate object.

    :type  format: str
    :param format: The output format, either 'tree' or 'flat'.
                   The 'tree' format is the same as returned
                   returned by SSLSocket.getpeercert and
                   accepted by ssl.match_hostname. The 'flat' format
                   uses just one level of key-value pairs, and all
                   values are strings, and is accepted by Prometheus
                   info metrics.
```

--------------------------------

### Manage Client Authorization Keys

Source: https://onionservices.torproject.org/apps/web/onionprobe/standalone

Commands to generate, restart, show, and remove client authorization keys for service protection.

```bash
./onionprobe-monitor restart tor
```

```bash
./onionprobe-monitor showkeys
```

```bash
./onionprobe-monitor showkeys myuser
```

```bash
./onionprobe-monitor removekeys myuser
./onionprobe-monitor restart tor
```

--------------------------------

### Get Endpoint by Public Key

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Searches the configured endpoints to find a match for the provided Onion Service public key.

```python
def get_endpoint_by_pubkey(self, pubkey):
    """
    Get an endpoint configuration given an Onion Service pubkey.

    :type  pubkey: str
    :param pubkey: Onion Service pubkey

    :rtype: tuple or False
    :return: Endpoint name and configuration if a match is found.
             False otherwise.
    """

    endpoints = self.get_config('endpoints')

    for name in endpoints:
        if self.get_pubkey_from_address(endpoints[name]['address']) == pubkey:
            return (name, endpoints[name])

    return False
```

--------------------------------

### Get Certificate Validity Interval in Python

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Retrieves the validity interval of an X.509 certificate, automatically selecting the appropriate UTC or legacy attributes based on the environment.

```python
def get_cert_validity_interval(self, cert, timestamp=False):
    """
    Handle deprecation transition on certificate validity.

    cryptography.x509.Certificate.not_valid_before and
    cryptography.x509.Certificate.not_valid_after deprecated in favor of
    cryptography.x509.Certificate.not_valid_before_utc and
    cryptography.x509.Certificate.not_valid_after_utc

    Details:

    https://gitlab.torproject.org/tpo/onion-services/onionprobe/-/issues/92
    https://cryptography.io/en/44.0.2/x509/reference/#cryptography.x509.Certificate.not_valid_before
    https://cryptography.io/en/44.0.2/x509/reference/#cryptography.x509.Certificate.not_valid_before_utc
    https://cryptography.io/en/44.0.2/x509/reference/#cryptography.x509.Certificate.not_valid_after
    https://cryptography.io/en/44.0.2/x509/reference/#cryptography.x509.Certificate.not_valid_after_utc

    :type  cert: cryptography.x509.Certificate
    :param cert: The X.509 Certificate object.

    :type  timestamp: bol
    :param timestamp: Whether the return values should be converted to a timestamp.
                      Defaults to False.

    :rtype: typle
    :return: Pair of certificate not valid before and not valid after timestamps.

    """

    if 'not_valid_before_utc' in dir(cert):
        not_valid_before = cert.not_valid_before_utc
        not_valid_after  = cert.not_valid_after_utc
    else:
        not_valid_before = cert.not_valid_before
        not_valid_after  = cert.not_valid_after

    if timestamp:
        return (not_valid_before.timestamp(), not_valid_after.timestamp())

    return (not_valid_before, not_valid_after)
```

--------------------------------

### Initialize Onionprobe configuration and subsystems

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

The OnionprobeInit class manages configuration from environment variables, files, and command-line arguments. The initialize method sets up logging, Tor daemon connectivity, and metrics exporters.

```python
class OnionprobeInit:
    #
    # Initialization logic
    #

    def __init__(self, args):
        """
        Onionprobe class constructor.

        Setup instance configuration.

        Handles command-line parameters.

        :type  args: dict
        :param args: Instance arguments.
        """

        self.args = args
        self.data = []

        # Environment variable handling
        if 'ONIONPROBE_CONFIG' in os.environ and os.environ['ONIONPROBE_CONFIG'] != '':
            args.config = os.environ['ONIONPROBE_CONFIG']

        # Config file handling
        if args.config is not None:
            if os.path.exists(args.config):
                with open(args.config, 'r') as config:
                    self.config = yaml.load(config, yaml.CLoader)
            else:
                raise FileNotFoundError(args.config)
        else:
            self.config = {}

        # Endpoints argument handling
        if args.endpoints is not None:
            import urllib.parse

            if 'endpoints' not in self.config:
                self.config['endpoints'] = {}

            for endpoint in args.endpoints:
                try:
                    url          = urllib.parse.urlparse(endpoint)
                    default_port = '443' if url.scheme == 'https' else '80'

                    # Check if only the onion address was provided, without protocol information
                    if url.path == endpoint:
                        url = urllib.parse.urlparse('http://' + endpoint)

                    # Remove port from the address information
                    if url.port is not None:
                        (address, port) = tuple(url.netloc.split(':'))
                    else:
                        address = url.netloc

                    self.config['endpoints'][endpoint] = {
                        'address' : address,
                        'protocol': url.scheme,
                        'port'    : str(url.port) if url.port is not None else default_port,
                        'paths'   : [{
                                        'path': url.path if url.path != '' else '/',
                                },
                            ],
                        }

                except ValueError as e:
                    self.log('Invalid URL {}, skipping.'.format(endpoint))

                    continue

        from .config import config

        # Handle all other arguments
        for argument in config:
            if argument == 'endpoints':
                continue

            value = getattr(args, argument)

            if value is not None and value != config[argument]['default']:
                self.config[argument] = value

    def initialize(self):
        """
        Onionprobe initialization procedures

        Initializes all Onionprobe subsystems, like the random number generator,
        logging, metrics and a Tor daemon instance.

        :rtype: bol
        :return: True if initialization is successful, False on error
        """

        # Initializes the random number generator
        random.seed()

        # Initializes logging
        if self.initialize_logging() is False:
            return False

        # Initializes the Tor daemon
        if self.initialize_tor() is False:
            return False

        # Authenticate with the Tor daemon
        if self.initialize_tor_auth() is False:
            return False

        # Initialize Tor event listeners
        self.initialize_listeners()

        # Initialize the Prometheus exporter
        if self.get_config('prometheus_exporter'):
            # Enforce continuous run
            self.config['loop'] = True

            if self.initialize_prometheus_exporter() is False:
                return False

        # Initialize metrics
        self.initialize_metrics()

        self.log('Onionprobe is initialized. Hit Ctrl-C to interrupt it.')

        return True
```

--------------------------------

### Run Onionprobe from Command Line

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Entry point for executing the probe using command-line configuration arguments.

```python
def run_from_cmdline():
    """
    Run Onionprobe getting arguments from the command line.
    """

    from .config import cmdline

    run(cmdline())
```

--------------------------------

### new_circuit(stream)

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Sets up a fresh Tor circuit for new streams by closing the existing circuit and attaching a new one to the provided stream.

```APIDOC
## new_circuit(stream)

### Description
Setup a fresh Tor circuit for new streams. This method removes the old circuit if it exists and attaches the new circuit to the provided stream.

### Parameters
- **stream** (stem.response.events.StreamEvent) - Required - Stream event to attach to the new circuit.
```

--------------------------------

### Run Onionprobe from source

Source: https://onionservices.torproject.org/apps/web/onionprobe/installation

Executes the Onionprobe binary from the working directory.

```bash
./onionprobe
```

--------------------------------

### Execute Onionprobe from Arguments

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Initializes and runs the probe instance based on provided arguments, including signal registration and exception handling.

```python
def run(args):
    """
    Run Onionprobe from arguments

    :type  args: dict
    :param args: Instance arguments.
    """

    # Register signal handling
    #signal.signal(signal.SIGINT, finish_handler)
    signal.signal(signal.SIGTERM, finish_handler)

    # Exit status (shell convention means 0 is success, failure otherwise)
    status = 0

    # Dispatch
    try:
        probe = Onionprobe(args)

        if probe.initialize() is not False:
            status = 0 if probe.run() else 1
        else:
            status = 1

            print('Error: could not initialize')

    # Handle user interruption
    # See https://stackoverflow.com/questions/21120947/catching-keyboardinterrupt-in-python-during-program-shutdown
    except KeyboardInterrupt as e:
        probe.log('Stopping Onionprobe due to user request...')

    except FileNotFoundError as e:
        status = 1

        print('File not found: ' + str(e))

    except Exception as e:
        status = 1

        print(repr(e))

    finally:
        if 'probe' in locals():
            probe.close()

        finish(status)
```

--------------------------------

### initialize_logging()

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Initializes the Onionprobe logging subsystem based on the configured log level.

```APIDOC
## initialize_logging()

### Description
Initializes the Onionprobe logging subsystem. It configures the logging level and sets up the Stem logger.

### Returns
- **bool** - True if initialization is successful, False on error.
```

--------------------------------

### Generate command line arguments for the configuration compiler

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Initializes an argument parser for the configuration compiler, including logic to locate the default configuration template.

```python
def cmdline_parser_compiler(default_source=None):
    """
    Generate command line arguments for the configuration compiler

    :rtype: argparse.ArgumentParser
    :return: The parser object
    """

    description = 'Generates an Onionprobe config file from ' + default_source
    parser      = argparse.ArgumentParser(
                    description=description,
                    formatter_class=argparse.RawDescriptionHelpFormatter,
                  )

    # Try to use the configs/ folder as the default config_template (will match
    # when running directly from the Onionprobe repository or from the python
    # package)
    config_template = os.path.normpath(os.path.join(basepath, 'configs', 'tor.yaml'))

    # Fallback config_template to /etc/onionprobe
    if not os.path.exists(config_template):
        config_template = os.path.normpath(os.path.join(os.sep, 'etc', 'onionprobe', 'tor.yaml'))
```

--------------------------------

### Initialize Onionprobe instance

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Constructor for the Onionprobe class that sets up configuration from arguments, environment variables, and YAML files.

```python
def __init__(self, args):
    """
    Onionprobe class constructor.

    Setup instance configuration.

    Handles command-line parameters.

    :type  args: dict
    :param args: Instance arguments.
    """

    self.args = args
    self.data = []

    # Environment variable handling
    if 'ONIONPROBE_CONFIG' in os.environ and os.environ['ONIONPROBE_CONFIG'] != '':
        args.config = os.environ['ONIONPROBE_CONFIG']

    # Config file handling
    if args.config is not None:
        if os.path.exists(args.config):
            with open(args.config, 'r') as config:
                self.config = yaml.load(config, yaml.CLoader)
        else:
            raise FileNotFoundError(args.config)
    else:
        self.config = {}

    # Endpoints argument handling
    if args.endpoints is not None:
        import urllib.parse

        if 'endpoints' not in self.config:
            self.config['endpoints'] = {}

        for endpoint in args.endpoints:
            try:
                url          = urllib.parse.urlparse(endpoint)
                default_port = '443' if url.scheme == 'https' else '80'

                # Check if only the onion address was provided, without protocol information
                if url.path == endpoint:
                    url = urllib.parse.urlparse('http://' + endpoint)

                # Remove port from the address information
                if url.port is not None:
                    (address, port) = tuple(url.netloc.split(':'))
                else:
                    address = url.netloc

                self.config['endpoints'][endpoint] = {
                    'address' : address,
                    'protocol': url.scheme,
                    'port'    : str(url.port) if url.port is not None else default_port,
                    'paths'   : [{
                                    'path': url.path if url.path != '' else '/',
                            },
                        ],
                    }

            except ValueError as e:
                self.log('Invalid URL {}, skipping.'.format(endpoint))

                continue

    from .config import config

    # Handle all other arguments
    for argument in config:
        if argument == 'endpoints':
            continue

        value = getattr(args, argument)

        if value is not None and value != config[argument]['default']:
            self.config[argument] = value
```

--------------------------------

### Build and wait execution

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Executes the configuration build process and pauses execution for a specified duration if configured.

```python
def build_and_wait(self):
        """
        Build Onionprobe configs, then wait.
        """

        self.build_onionprobe_config()

        if self.wait != 0:
            import time

            print('Waiting %s seconds...' % (self.wait))
            time.sleep(self.wait)
```

--------------------------------

### Register changes

Source: https://onionservices.torproject.org/apps/web/onionprobe/development

Update changelog, commit, and tag the release.

```bash
$EDITOR ChangeLog

```

```bash
git diff # review
git commit -a -m "Feat: Onionprobe $ONIONPROBE_VERSION"
git tag -s $ONIONPROBE_VERSION -m "Onionprobe $ONIONPROBE_VERSION"

```

--------------------------------

### initialize_metrics()

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Initializes the internal metrics subsystem using Prometheus metrics.

```APIDOC
## initialize_metrics()

### Description
Initializes the metrics subsystem. This method ensures Prometheus metrics are available for internal use, even if the Prometheus exporter is not explicitly enabled.
```

--------------------------------

### Launch Tor Process

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Launches a built-in Tor process with custom configuration and bootstrap logging.

```python
def launch_tor(self):
    """
    Launch a built-in Tor process

    See https://stem.torproject.org/tutorials/to_russia_with_love.html
        https://stem.torproject.org/api/process.html

    """

    # Check if the tor executable is available
    if shutil.which('tor') is None:
        self.log('Cannot find the tor executable. Is it installed?', 'critical')

        return False

    from stem.util import term

    # Helper function to print bootstrap lines
    def print_bootstrap_lines(line):
        level = self.get_config('log_level')

        if '[debug]' in line:
            self.log(term.format(line), 'debug')
        elif '[info]' in line:
            self.log(term.format(line), 'debug')
        elif '[notice]' in line:
            self.log(term.format(line), 'debug')
        elif '[warn]' in line:
            self.log(term.format(line), 'warning')
        elif '[err]' in line:
            self.log(term.format(line), 'error')

    try:
        self.log('Initializing Tor process (might take a while to bootstrap)...')

        tor_address         = self.get_config('tor_address')
        control_password    = self.get_config('control_password', self.gen_control_password())
        metrics_port        = self.get_config('metrics_port')
        metrics_port_policy = self.get_config('metrics_port_policy')
        config              = {
            'SocksPort'            : tor_address + ':' + str(self.get_config('socks_port')),
            'ControlPort'          : tor_address + ':' + str(self.get_config('control_port')),
            'HashedControlPassword': self.hash_password(control_password),
            'CircuitStreamTimeout' : str(self.get_config('circuit_stream_timeout')),
            }

        # Log config
        #config['Log'] = [
        #    'DEBUG  stdout',
        #    'INFO   stdout',
        #    'NOTICE stdout',
        #    'WARN   stdout',
        #    'ERR    stdout',
        #    ]

        if metrics_port is not None and metrics_port != '' and metrics_port != 0:
            config['MetricsPort'] = str(metrics_port)

        if metrics_port_policy is not None and metrics_port_policy != '':
            config['MetricsPortPolicy'] = str(metrics_port_policy)

        self.tor = stem.process.launch_tor_with_config(
                config           = config,
                init_msg_handler = print_bootstrap_lines,
                )

    except OSError as e:
        self.log(e, 'error')

        return False
```

--------------------------------

### OnionprobeConfigCompiler.build_and_wait

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Executes the configuration build process and waits for the specified interval if configured.

```APIDOC
## build_and_wait()

### Description
Builds the Onionprobe configuration files and pauses execution for the duration defined in the wait parameter.
```

--------------------------------

### Build Onionprobe configuration

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Generates and writes Onionprobe-compatible YAML configuration files for each database defined in the class attributes.

```python
def build_onionprobe_config(self):
        """
        Build an Onionprobe config.

        Writes an Onionprobe-compatible configuration file for each database
        listed in self.databases attribute.

        The Onion Service endpoints are generated from the
        build_endpoints_config() methods. To be effective, it's required that
        classes inheriting from this base class to implement the
        build_endpoints_configs() method.

        The filenames ared derived from the database names (each key from the
        self.databases attribute).
        """

        for database in self.databases:
            try:
                print('Building the list of endpoints for database %s...' % (database))

                # Build list of endpoints
                endpoints = self.build_endpoints_config(database)

                # Create a new config using the default as base
                new_config = dict(self.config)

                # Replace the endpoints
                new_config['endpoints'] = endpoints

                # Build the output path
                output_folder = os.path.normpath(os.path.join(self.output_folder, database + '.yaml'))

                # Save
                with open(output_folder, 'w') as output:
                    print('Saving the generated config for database %s into %s...' % (database, output_folder))

                    output.write(yaml.dump(new_config))

            except Exception as e:
                print(e)
```

--------------------------------

### Initialize Onionprobe Subsystems

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Initializes random number generation, logging, Tor daemon, and metrics. Returns a boolean indicating success or failure.

```python
def initialize(self):
    """
    Onionprobe initialization procedures

    Initializes all Onionprobe subsystems, like the random number generator,
    logging, metrics and a Tor daemon instance.

    :rtype: bol
    :return: True if initialization is successful, False on error
    """

    # Initializes the random number generator
    random.seed()

    # Initializes logging
    if self.initialize_logging() is False:
        return False

    # Initializes the Tor daemon
    if self.initialize_tor() is False:
        return False

    # Authenticate with the Tor daemon
    if self.initialize_tor_auth() is False:
        return False

    # Initialize Tor event listeners
    self.initialize_listeners()

    # Initialize the Prometheus exporter
    if self.get_config('prometheus_exporter'):
        # Enforce continuous run
        self.config['loop'] = True

        if self.initialize_prometheus_exporter() is False:
            return False

    # Initialize metrics
    self.initialize_metrics()

    self.log('Onionprobe is initialized. Hit Ctrl-C to interrupt it.')

    return True
```

--------------------------------

### build_url(config, path=None)

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Constructs a full Onion Service URL based on the provided configuration dictionary and an optional path.

```APIDOC
## build_url(config, path=None)

### Description
Builds an Onion Service URL to be probed using the provided endpoint configuration and an optional path.

### Parameters
- **config** (dict) - Required - Endpoint configuration containing address, protocol, and port information.
- **path** (str) - Optional - The path to be appended to the base URL.

### Returns
- **str** - The complete Onion Service URL.
```

--------------------------------

### run_from_cmdline()

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Executes the Onionprobe instance by automatically parsing arguments from the command line.

```APIDOC
## run_from_cmdline()

### Description
Run Onionprobe getting arguments from the command line.
```

--------------------------------

### Initialize Tor control connection

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/onionprobe

Establishes a connection to the Tor controller, supporting optional local Tor process launching and IP address resolution.

```python
def initialize_tor(self):
    """
    Initialize Tor control connection

    :rtype: bol
    :return: True if initialization is successful, False on error
    """

    control_address = self.get_config('tor_address')
    control_port    = self.get_config('control_port')

    # Ensure control_address is an IP address, which is the
    # only type currently supported by stem
    #
    # Right now only IPv4 is supported
    import socket
    control_address = socket.gethostbyname(control_address)

    if self.get_config('launch_tor'):
        if self.launch_tor() is False:
            self.log("Error initializing Tor", "critical")

            return False

    try:
        self.controller = stem.control.Controller.from_port(
                address=control_address,
                port=control_port)

    except stem.SocketError as exc:
        self.log("Unable to connect to tor on port 9051: %s" % exc, "critical")

        return False

    return True
```

--------------------------------

### Build endpoints configuration

Source: https://onionservices.torproject.org/apps/web/onionprobe/api/helpers

Parses a database of onion sites to generate a structured dictionary of endpoints. Requires a valid database path or URL and handles CSV input.

```python
        with open(self.databases[database], 'r') as result:
            data = csv.DictReader(result.readlines())

    else:
        try:
            print('Fetching remote list of %s database endpoints from %s...' % (database, self.databases[database]))

            result = requests.get(self.databases[database])
            data   = csv.DictReader(StringIO(result.text))

        except Exception as e:
            # Log the exception
            print(repr(e))

            # Some error happened: do not proceed generating the config
            exit(1)

    # Parse the database and convert it to the Onionprobe endpoints format
    for item in data:
        print('Processing %s...' % (item['site_name']))

        url      = urllib.parse.urlparse(item['onion_url'])
        address  = url.netloc
        protocol = url.scheme if url.scheme != '' else 'http'
        port     = 80 if protocol == 'http' else 443
        paths    = [{
            'path'            : url.path if url.path != '' else '/',
            'allowed_statuses': [ 200 ],
            }]

        # Append to the endpoints dictionary
        if item['site_name'] not in endpoints:
            endpoints[item['site_name']] = {
                    'address' : address,
                    'protocol': protocol,
                    'port'    : port,
                    'paths'   : paths,
                    }

    return endpoints
```

--------------------------------

### Update version files

Source: https://onionservices.torproject.org/apps/web/onionprobe/development

Update project files with the new version.

```bash
dch -i # debian/changelog
$EDITOR packages/onionprobe/config.py
$EDITOR docker-compose.yaml
$EDITOR setup.cfg

```
Tokens:
Raw