Skip to content

Onionprobe package

Onionprobe: test and monitoring for Onion Services.

app

Onionprobe

Bases: OnionprobeInit, OnionprobeConfig, OnionprobeLogger, OnionprobeTime, OnionprobeTor, OnionprobeDescriptor, OnionprobeMetrics, OnionprobeProber, OnionprobeHTTP, OnionprobeTLS, OnionprobeCertificate, OnionprobeTeardown, OnionprobeMain

Onionprobe class to test and monitor Tor Onion Services

Source code in packages/onionprobe/app.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class Onionprobe(
        # Inherit from subsystems
        OnionprobeInit,
        OnionprobeConfig,
        OnionprobeLogger,
        OnionprobeTime,
        OnionprobeTor,
        OnionprobeDescriptor,
        OnionprobeMetrics,
        OnionprobeProber,
        OnionprobeHTTP,
        OnionprobeTLS,
        OnionprobeCertificate,
        OnionprobeTeardown,
        OnionprobeMain,
        ):
    """
    Onionprobe class to test and monitor Tor Onion Services
    """

finish(status=0)

Stops Onionprobe

Parameters:

Name Type Description Default
status

Exit status code.

 0
Source code in packages/onionprobe/app.py 
60
61
62
63
64
65
66
67
68
69
70
71
def finish(status=0):
    """
    Stops Onionprobe

    :type  status: int
    :param status: Exit status code.
    """

    try:
        sys.exit(status)
    except SystemExit:
        os._exit(status)

finish_handler(signal, frame)

Wrapper around finish() for handling system signals

Parameters:

Name Type Description Default
signal

Signal number.

 required
frame

Current stack frame.

 required
Source code in packages/onionprobe/app.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def finish_handler(signal, frame):
    """
    Wrapper around finish() for handling system signals

    :type  signal: int
    :param signal: Signal number.

    :type  frame: object
    :param frame: Current stack frame.
    """

    print('Signal received, stopping Onionprobe..')

    finish(1)

run(args)

Run Onionprobe from arguments

Parameters:

Name Type Description Default
args

Instance arguments.

 required
Source code in packages/onionprobe/app.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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)

run_from_cmdline()

Run Onionprobe getting arguments from the command line.

Source code in packages/onionprobe/app.py 
135
136
137
138
139
140
141
142
def run_from_cmdline():
    """
    Run Onionprobe getting arguments from the command line.
    """

    from .config import cmdline

    run(cmdline())

certificate

OnionprobeCertificate

Onionprobe class with X.509 Certificate methods.

Source code in packages/onionprobe/certificate.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
class OnionprobeCertificate:
    """
    Onionprobe class with X.509 Certificate methods.
    """

    def get_dns_alt_names_from_cert(self, cert, format='tuple'):
        """
        Get the DNS names from a X.509 certificate's SubjectAltName extension.

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

        :type  format: str
        :param format: The output format, either 'list' or 'tuple' in the
                       same format returned by SSLSocket.getpeercert and
                       accepted by ssl.match_hostname.

        :rtype: list or tuple
        :return: The list or tuple with the certificate's DNS Subject
                 Alternative Names.

        """

        dns_alt_names = cert.extensions.get_extension_for_oid(
                oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME
                ).value.get_values_for_type(DNSName)

        if format == 'tuple':
            dns_alt_names = tuple(('DNS', item) for item in dns_alt_names)

        return dns_alt_names

    def get_cert_rdns(self, cert, field = 'issuer', format = 'tuple'):
        """
        Get the Relative Distinguished Names (RDNs) from a given X.509
        certificate field like issuer or subject.

        :type field: str
        :param field: The name of the X.509 certificate field
                      ('issuer' or 'subject').

        :type  format: str
        :param format: The output format, either 'list' or 'tuple' in the
                       same format returned by SSLSocket.getpeercert and
                       accepted by ssl.match_hostname.

        :rtype: dict or tuple
        :return: The dict or tuple with the certificate's DNS Subject
                 Alternative Names.

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

        """

        items = {}

        for item in getattr(cert, field):
            name = item.oid._name

            if name not in items:
                items[name] = []

            items[name].append(item.value)

        if format == 'dict':
            return items

        result = []

        for name in items:
            result.append(tuple((name, item) for item in items[name]))

        return tuple(result)

    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.

        :rtype: dict
        :return: Dictionary with basic certificate information in the same
                 format returned by SSLSocket.getpeercert and accepted by
                 ssl.match_hostname, additionally with certificate
                 fingerprints.

        """

        # Date format is the same from ssl.cert_time_to_seconds
        date_format = '%b %d %H:%M:%S %Y %Z'

        # The info dictionary
        info = {
                'issuer'           : self.get_cert_rdns(cert, 'issuer'),
                'subject'          : self.get_cert_rdns(cert, 'subject'),
                'subjectAltName'   : self.get_dns_alt_names_from_cert(cert),

                # Convert to aware datetime formats since
                # cryptography.x509.Certificate uses naive objects by default
                'notAfter'         : cert.not_valid_after.replace(
                    tzinfo=timezone.utc).strftime(date_format),
                'notBefore'        : cert.not_valid_before.replace(
                    tzinfo=timezone.utc).strftime(date_format),

                'serialNumber'     : str(cert.serial_number),
                'version'          : int(str(cert.version).replace('Version.v', '')),

                'fingerprintSHA1'  : cert.fingerprint(hashes.SHA1()).hex(':').upper(),
                'fingerprintSHA256': cert.fingerprint(hashes.SHA256()).hex(':').upper(),
        }

        if format == 'flat':
            info['version']        = str(info['version'])
            info['issuer']         = cert.issuer.rfc4514_string()
            info['subject']        = cert.subject.rfc4514_string()
            info['subjectAltName'] = ' '.join(self.get_dns_alt_names_from_cert(cert, 'list'))

        return info

    def get_certificate_expiration(self, cert):
        """
        Get the number of seconds remaining before a X.509 certificate expires,
        or the number of seconds passed since it's expiration.

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

        :rtype: int
        :return: Number of seconding remaining before the certificate
                 expiration (if positive) or the number of seconds passed since the
                 expiration (if negative).

        """

        not_valid_after = cert.not_valid_after.replace(tzinfo=timezone.utc).timestamp()
        now             = datetime.now(timezone.utc).timestamp()

        return int(not_valid_after - now)

    def get_certificate(self, endpoint, config, tls):
        """
        Get the certificate information from a TLS connection.

        :type  endpoint: str
        :param endpoint: The endpoint name from the 'endpoints' instance config.

        :type  config: dict
        :param config: Endpoint configuration

        :type  tls: ssl.SSLSocket
        :param tls: The TLS socket connection to the endpoint.

        :rtype: cryptography.x509.Certificate or False
        :return: The X.509 certificate object on success.
                 False on error.

        """

        try:
            # We can't rely on ssl.getpeercert() if the certificate wasn't validated
            #cert_info = tls.getpeercert()

            self.log('Retrieving certificate information for {} on port {}'.format(
                    config['address'], config['port']))

            der_cert         = tls.getpeercert(binary_form=True)
            pem_cert         = ssl.DER_cert_to_PEM_cert(der_cert)
            cert             = load_pem_x509_certificate(bytes(pem_cert, 'utf-8'))
            result           = cert
            not_valid_before = cert.not_valid_before.timestamp()
            not_valid_after  = cert.not_valid_after.timestamp()
            info             = self.get_cert_info(cert)
            expiry           = self.get_certificate_expiration(cert)
            match_hostname   = 1
            labels           = {
                    'name'    : endpoint,
                    'address' : config['address'],
                    'port'    : config['port'],
                    }

            try:
                match = ssl.match_hostname(info, config['address'])

            except ssl.CertificateError as e:
                match_hostname = 0

            self.info_metric('onion_service_certificate', self.get_cert_info(cert, 'flat'), labels)

            self.set_metric('onion_service_certificate_not_valid_before_timestamp_seconds',
                    not_valid_before, labels)
            self.set_metric('onion_service_certificate_not_valid_after_timestamp_seconds',
                    not_valid_after, labels)
            self.set_metric('onion_service_certificate_expiry_seconds', expiry,         labels)
            self.set_metric('onion_service_certificate_match_hostname', match_hostname, labels)

            message = 'Certificate for {address} on {port} has subject: {subject}; ' + \
                      'issuer: {issuer}; serial number: {serial_number}; version: {version}; ' + \
                      'notBefore: {not_before}; notAfter: {not_after}; SHA256 fingerprint: ' + \
                      '{fingerprint}'

            self.log(message.format(
                address       = config['address'],
                port          = config['port'],
                subject       = cert.subject.rfc4514_string(),
                issuer        = cert.issuer.rfc4514_string(),
                serial_number = info['serialNumber'],
                version       = str(info['version']),
                not_before    = info['notBefore'],
                not_after     = info['notAfter'],
                fingerprint   = info['fingerprintSHA256'],
                ))

            if expiry <= 0:
                self.log('The certificate for {address} on port {port} expired {days} days ago'.format(
                    address = config['address'],
                    port    = config['port'],
                    days    = str(int(-1 * expiry / 86400)),
                    ), 'error')

        except Exception as e:
            result    = False
            exception = 'generic_error'

            self.log(e, 'error')

        finally:
            return result

get_cert_info(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.

Parameters:

Name Type Description Default
cert

The X.509 Certificate object.

 required
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.

 'tree'

Returns:

Type Description
dict

Dictionary with basic certificate information in the same format returned by SSLSocket.getpeercert and accepted by ssl.match_hostname, additionally with certificate fingerprints.
Source code in packages/onionprobe/certificate.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
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.

    :rtype: dict
    :return: Dictionary with basic certificate information in the same
             format returned by SSLSocket.getpeercert and accepted by
             ssl.match_hostname, additionally with certificate
             fingerprints.

    """

    # Date format is the same from ssl.cert_time_to_seconds
    date_format = '%b %d %H:%M:%S %Y %Z'

    # The info dictionary
    info = {
            'issuer'           : self.get_cert_rdns(cert, 'issuer'),
            'subject'          : self.get_cert_rdns(cert, 'subject'),
            'subjectAltName'   : self.get_dns_alt_names_from_cert(cert),

            # Convert to aware datetime formats since
            # cryptography.x509.Certificate uses naive objects by default
            'notAfter'         : cert.not_valid_after.replace(
                tzinfo=timezone.utc).strftime(date_format),
            'notBefore'        : cert.not_valid_before.replace(
                tzinfo=timezone.utc).strftime(date_format),

            'serialNumber'     : str(cert.serial_number),
            'version'          : int(str(cert.version).replace('Version.v', '')),

            'fingerprintSHA1'  : cert.fingerprint(hashes.SHA1()).hex(':').upper(),
            'fingerprintSHA256': cert.fingerprint(hashes.SHA256()).hex(':').upper(),
    }

    if format == 'flat':
        info['version']        = str(info['version'])
        info['issuer']         = cert.issuer.rfc4514_string()
        info['subject']        = cert.subject.rfc4514_string()
        info['subjectAltName'] = ' '.join(self.get_dns_alt_names_from_cert(cert, 'list'))

    return info

get_cert_rdns(cert, field='issuer', format='tuple')

Get the Relative Distinguished Names (RDNs) from a given X.509 certificate field like issuer or subject.

Parameters:

Name Type Description Default
field str

The name of the X.509 certificate field ('issuer' or 'subject').

 'issuer'
format

The output format, either 'list' or 'tuple' in the same format returned by SSLSocket.getpeercert and accepted by ssl.match_hostname.

 'tuple'
cert

The X.509 Certificate object.

 required

Returns:

Type Description
dict | tuple

The dict or tuple with the certificate's DNS Subject Alternative Names.
Source code in packages/onionprobe/certificate.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_cert_rdns(self, cert, field = 'issuer', format = 'tuple'):
    """
    Get the Relative Distinguished Names (RDNs) from a given X.509
    certificate field like issuer or subject.

    :type field: str
    :param field: The name of the X.509 certificate field
                  ('issuer' or 'subject').

    :type  format: str
    :param format: The output format, either 'list' or 'tuple' in the
                   same format returned by SSLSocket.getpeercert and
                   accepted by ssl.match_hostname.

    :rtype: dict or tuple
    :return: The dict or tuple with the certificate's DNS Subject
             Alternative Names.

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

    """

    items = {}

    for item in getattr(cert, field):
        name = item.oid._name

        if name not in items:
            items[name] = []

        items[name].append(item.value)

    if format == 'dict':
        return items

    result = []

    for name in items:
        result.append(tuple((name, item) for item in items[name]))

    return tuple(result)

get_certificate(endpoint, config, tls)

Get the certificate information from a TLS connection.

Parameters:

Name Type Description Default
endpoint

The endpoint name from the 'endpoints' instance config.

 required
config

Endpoint configuration

 required
tls

The TLS socket connection to the endpoint.

 required

Returns:

Type Description
cryptography.x509.Certificate | False

The X.509 certificate object on success. False on error.
Source code in packages/onionprobe/certificate.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
def get_certificate(self, endpoint, config, tls):
    """
    Get the certificate information from a TLS connection.

    :type  endpoint: str
    :param endpoint: The endpoint name from the 'endpoints' instance config.

    :type  config: dict
    :param config: Endpoint configuration

    :type  tls: ssl.SSLSocket
    :param tls: The TLS socket connection to the endpoint.

    :rtype: cryptography.x509.Certificate or False
    :return: The X.509 certificate object on success.
             False on error.

    """

    try:
        # We can't rely on ssl.getpeercert() if the certificate wasn't validated
        #cert_info = tls.getpeercert()

        self.log('Retrieving certificate information for {} on port {}'.format(
                config['address'], config['port']))

        der_cert         = tls.getpeercert(binary_form=True)
        pem_cert         = ssl.DER_cert_to_PEM_cert(der_cert)
        cert             = load_pem_x509_certificate(bytes(pem_cert, 'utf-8'))
        result           = cert
        not_valid_before = cert.not_valid_before.timestamp()
        not_valid_after  = cert.not_valid_after.timestamp()
        info             = self.get_cert_info(cert)
        expiry           = self.get_certificate_expiration(cert)
        match_hostname   = 1
        labels           = {
                'name'    : endpoint,
                'address' : config['address'],
                'port'    : config['port'],
                }

        try:
            match = ssl.match_hostname(info, config['address'])

        except ssl.CertificateError as e:
            match_hostname = 0

        self.info_metric('onion_service_certificate', self.get_cert_info(cert, 'flat'), labels)

        self.set_metric('onion_service_certificate_not_valid_before_timestamp_seconds',
                not_valid_before, labels)
        self.set_metric('onion_service_certificate_not_valid_after_timestamp_seconds',
                not_valid_after, labels)
        self.set_metric('onion_service_certificate_expiry_seconds', expiry,         labels)
        self.set_metric('onion_service_certificate_match_hostname', match_hostname, labels)

        message = 'Certificate for {address} on {port} has subject: {subject}; ' + \
                  'issuer: {issuer}; serial number: {serial_number}; version: {version}; ' + \
                  'notBefore: {not_before}; notAfter: {not_after}; SHA256 fingerprint: ' + \
                  '{fingerprint}'

        self.log(message.format(
            address       = config['address'],
            port          = config['port'],
            subject       = cert.subject.rfc4514_string(),
            issuer        = cert.issuer.rfc4514_string(),
            serial_number = info['serialNumber'],
            version       = str(info['version']),
            not_before    = info['notBefore'],
            not_after     = info['notAfter'],
            fingerprint   = info['fingerprintSHA256'],
            ))

        if expiry <= 0:
            self.log('The certificate for {address} on port {port} expired {days} days ago'.format(
                address = config['address'],
                port    = config['port'],
                days    = str(int(-1 * expiry / 86400)),
                ), 'error')

    except Exception as e:
        result    = False
        exception = 'generic_error'

        self.log(e, 'error')

    finally:
        return result

get_certificate_expiration(cert)

Get the number of seconds remaining before a X.509 certificate expires, or the number of seconds passed since it's expiration.

Parameters:

Name Type Description Default
cert

The X.509 Certificate object.

 required

Returns:

Type Description
int

Number of seconding remaining before the certificate expiration (if positive) or the number of seconds passed since the expiration (if negative).
Source code in packages/onionprobe/certificate.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def get_certificate_expiration(self, cert):
    """
    Get the number of seconds remaining before a X.509 certificate expires,
    or the number of seconds passed since it's expiration.

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

    :rtype: int
    :return: Number of seconding remaining before the certificate
             expiration (if positive) or the number of seconds passed since the
             expiration (if negative).

    """

    not_valid_after = cert.not_valid_after.replace(tzinfo=timezone.utc).timestamp()
    now             = datetime.now(timezone.utc).timestamp()

    return int(not_valid_after - now)

get_dns_alt_names_from_cert(cert, format='tuple')

Get the DNS names from a X.509 certificate's SubjectAltName extension.

Parameters:

Name Type Description Default
cert

The X.509 Certificate object.

 required
format

The output format, either 'list' or 'tuple' in the same format returned by SSLSocket.getpeercert and accepted by ssl.match_hostname.

 'tuple'

Returns:

Type Description
list | tuple

The list or tuple with the certificate's DNS Subject Alternative Names.
Source code in packages/onionprobe/certificate.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def get_dns_alt_names_from_cert(self, cert, format='tuple'):
    """
    Get the DNS names from a X.509 certificate's SubjectAltName extension.

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

    :type  format: str
    :param format: The output format, either 'list' or 'tuple' in the
                   same format returned by SSLSocket.getpeercert and
                   accepted by ssl.match_hostname.

    :rtype: list or tuple
    :return: The list or tuple with the certificate's DNS Subject
             Alternative Names.

    """

    dns_alt_names = cert.extensions.get_extension_for_oid(
            oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME
            ).value.get_values_for_type(DNSName)

    if format == 'tuple':
        dns_alt_names = tuple(('DNS', item) for item in dns_alt_names)

    return dns_alt_names

config

OnionprobeConfig

Onionprobe class with configuration-related methods.

Source code in packages/onionprobe/config.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
class OnionprobeConfig:
    """
    Onionprobe class with configuration-related methods.
    """

    def get_config(self, item, default = None):
        """
        Helper to get instance configuration

        Retrieve a config parameter from the self.config object or use a
        default value as fallback

        :type  item: str
        :param item: Configuration item name

        :param default: Default config value to be used as a fallback if there's
                        no self.config[item] available.
                        Defaults to None

        :return: The configuration parameter value or the default fallback value.
        """

        if self.config is None:
            self.config = {}

        if item in self.config:
            return self.config[item]

        # Optionally override the default with an argument provided
        elif default is not None:
            self.config[item] = default

            return default

        return config[item]['default']

get_config(item, default=None)

Helper to get instance configuration

Retrieve a config parameter from the self.config object or use a default value as fallback

Parameters:

Name Type Description Default
item

Configuration item name

 required
default

Default config value to be used as a fallback if there's no self.config[item] available. Defaults to None

 None

Returns:

Type Description

The configuration parameter value or the default fallback value.
Source code in packages/onionprobe/config.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def get_config(self, item, default = None):
    """
    Helper to get instance configuration

    Retrieve a config parameter from the self.config object or use a
    default value as fallback

    :type  item: str
    :param item: Configuration item name

    :param default: Default config value to be used as a fallback if there's
                    no self.config[item] available.
                    Defaults to None

    :return: The configuration parameter value or the default fallback value.
    """

    if self.config is None:
        self.config = {}

    if item in self.config:
        return self.config[item]

    # Optionally override the default with an argument provided
    elif default is not None:
        self.config[item] = default

        return default

    return config[item]['default']

OnionprobeConfigCompiler

Base class to build Onionprobe configs from external sources of Onion Services

Source code in packages/onionprobe/config.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
class OnionprobeConfigCompiler:
    """Base class to build Onionprobe configs from external sources of Onion Services"""

    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))

    def build_endpoints_config(self, database):
        """
        Build the Onion Service endpoints dictionary.

        This method is only a placeholder.

        By default this method returns an empty dictionary as it's meant to be
        overriden by specific implementations inheriting from the
        OnionprobeConfigCompiler base class and where custom logic for
        extracting .onion endpoints from external databases should be located.

        :type database : str
        :param database: A database name from the databases dictionary. This
                         parameter allows accesing the URL of the external
                         database from the self.databases class attribute.

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

        return dict()

    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)

    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)

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

        """

        if self.loop is True:
            while True:
                self.build_and_wait()
        else:
            self.build_and_wait()

__init__(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.

Parameters:

Name Type Description Default
databases

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

 required
config_template

Configuration file path to be used as template

 required
output_folder

Output folder where configs are written

 required
Source code in packages/onionprobe/config.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
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))

build_and_wait()

Build Onionprobe configs, then wait.

Source code in packages/onionprobe/config.py 
483
484
485
486
487
488
489
490
491
492
493
494
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)

build_endpoints_config(database)

Build the Onion Service endpoints dictionary.

This method is only a placeholder.

By default this method returns an empty dictionary as it's meant to be overriden by specific implementations inheriting from the OnionprobeConfigCompiler base class and where custom logic for extracting .onion endpoints from external databases should be located.

Parameters:

Name Type Description Default
database

A database name from the databases dictionary. This parameter allows accesing the URL of the external database from the self.databases class attribute.

 required

Returns:

Type Description
dict

Onion Service endpoints in the format accepted by Onionprobe.
Source code in packages/onionprobe/config.py 
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
def build_endpoints_config(self, database):
    """
    Build the Onion Service endpoints dictionary.

    This method is only a placeholder.

    By default this method returns an empty dictionary as it's meant to be
    overriden by specific implementations inheriting from the
    OnionprobeConfigCompiler base class and where custom logic for
    extracting .onion endpoints from external databases should be located.

    :type database : str
    :param database: A database name from the databases dictionary. This
                     parameter allows accesing the URL of the external
                     database from the self.databases class attribute.

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

    return dict()

build_onionprobe_config()

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).

Source code in packages/onionprobe/config.py 
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
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)

compile()

Main compilation procedure.

Source code in packages/onionprobe/config.py 
496
497
498
499
500
501
502
503
504
505
506
def compile(self):
    """
    Main compilation procedure.

    """

    if self.loop is True:
        while True:
            self.build_and_wait()
    else:
        self.build_and_wait()

cmdline()

Evalutate the command line.

Returns:

Type Description
argparse.Namespace

Command line arguments.
Source code in packages/onionprobe/config.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
def cmdline():
    """
    Evalutate the command line.

    :rtype: argparse.Namespace
    :return: Command line arguments.
    """

    parser = cmdline_parser()
    args   = parser.parse_args()

    if args.config is None and args.endpoints is None:
        parser.print_usage()
        exit(1)

    return args

cmdline_compiler(default_source=None)

Evalutate the command line for the configuration compiler.

Returns:

Type Description
argparse.Namespace

Command line arguments.
Source code in packages/onionprobe/config.py 
578
579
580
581
582
583
584
585
586
587
588
589
def cmdline_compiler(default_source=None):
    """
    Evalutate the command line for the configuration compiler.

    :rtype: argparse.Namespace
    :return: Command line arguments.
    """

    parser = cmdline_parser_compiler(default_source)
    args   = parser.parse_args()

    return args

cmdline_parser()

Generate command line arguments

Returns:

Type Description
argparse.ArgumentParser

The parser object
Source code in packages/onionprobe/config.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def cmdline_parser():
    """
    Generate command line arguments

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

    epilog = """Examples:

      onionprobe -c configs/tor.yaml
      onionprobe -e http://2gzyxa5ihm7nsggfxnu52rck2vv4rvmdlkiu3zzui5du4xyclen53wid.onion
    """

    epilog += """\nAvailable metrics:
    """

    from .metrics import metrics

    for metric in metrics:
        item = metrics[metric].describe()[0]

        epilog += "\n  {}:\n        {}".format(item.name, item.documentation)

    description = 'Test and monitor onion services'
    parser      = argparse.ArgumentParser(
                    prog='onionprobe',
                    description=description,
                    epilog=epilog,
                    formatter_class=argparse.RawDescriptionHelpFormatter,
                  )

    parser.add_argument('-c', '--config', help="""
                        Read options from configuration file. All command line
                        parameters can be specified inside a YAML file.
                        Additional command line parameters override those set
                        in the configuration file.""".strip())

    parser.add_argument('-v', '--version', action='version', version='%(prog)s ' + onionprobe_version)

    for argument in sorted(config):
        if argument == 'endpoints':
            parser.add_argument('-e', '--endpoints', nargs='*', help='Add endpoints to the test list', metavar="ONION-ADDRESS1")

        else:
            config[argument]['type'] = type(config[argument]['default'])

            if not isinstance(config[argument]['default'], bool) and config[argument]['default'] != '':
                config[argument]['help'] += ' (default: %(default)s)'

            parser.add_argument('--' + argument, **config[argument])

    return parser

cmdline_parser_compiler(default_source=None)

Generate command line arguments for the configuration compiler

Returns:

Type Description
argparse.ArgumentParser

The parser object
Source code in packages/onionprobe/config.py 
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
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'))

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

    # Fallback output_folder to the current working directory
    if not os.path.exists(output_folder):
        output_folder = os.getcwd()

    parser.add_argument('-s', '--source',
            dest='source',
            default=default_source,
            help="Database source file or endpoint (default: %(default)s)")

    parser.add_argument('-t', '--config_template',
            dest='config_template',
            default=config_template,
            help="Configuration template to use (default %(default)s)")

    parser.add_argument('-o', '--output_folder',
            dest='output_folder',
            default=output_folder,
            help="Output folder where config should be saved (default: current working directory)")

    parser.add_argument('-w', '--wait',
            dest='wait',
            default=0,
            type=int,
            help="""Wait a number of seconds before exiting after writing the config.
                    Useful for a configurator container service tha should run periodically
                    (default: %(default)s)""".strip())

    parser.add_argument('-l', '--loop',
            dest='loop',
            action='store_true',
            default=False,
            help="""Whether to continuously generate configuration.
                    Useful when set in conjunction with --wait (default: %(default)s)""")

    parser.add_argument('-c', '--config_overrides',
            dest='config_overrides',
            default=None,
            nargs='*',
            help="Override configuration parameters in the form of param1=value1 ... paramN=valueN")

    return parser

descriptor

OnionprobeDescriptor

Onionprobe class with Tor descriptor-related methods.

Source code in packages/onionprobe/descriptor.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
class OnionprobeDescriptor:
    """
    Onionprobe class with Tor descriptor-related methods.
    """

    def get_pubkey_from_address(self, address):
        """
        Extract .onion pubkey from the address

        Leaves out the .onion domain suffix and any existing subdomains.

        :type  address: str
        :param address: Onion Service address

        :rtype: str
        :return: Onion Service public key
        """

        # Extract
        pubkey = address[0:-6].split('.')[-1]

        return pubkey

    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

    def parse_pow_params(self, inner_text, labels):
        """
        Parse the Proof of Work (PoW) parameters from a descriptor.

        :type  inner_text: str
        :param inner_text: The decrypted raw inner descriptor layer plaintext for the endpoint.

        :type  labels: dict
        :param labels: Metrics labels

        :rtype:  None
        :return: This method does not return any special value.
        """

        pow_params    = re.compile(r"^pow-params .*$", re.MULTILINE)
        pow_params_v1 = re.compile(
                r"^pow-params v1 (?P<seed>[^ ]*) (?P<effort>[0-9]*) (?P<expiration>[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2})$",
                re.MULTILINE)

        pow_parsed    = pow_params.search(inner_text)
        pow_parsed_v1 = pow_params_v1.search(inner_text)

        if pow_parsed:
            self.log("Proof of Work (PoW) params found in the descriptor")
            self.set_metric('onion_service_pow_enabled', 1, labels)
        else:
            self.log("Proof of Work (PoW) params not found in the descriptor")
            self.set_metric('onion_service_pow_enabled', 0, labels)

        if pow_parsed_v1:
            pow_data_v1 = pow_parsed_v1.groupdict()
            expiration  = int(datetime.datetime.fromisoformat(pow_data_v1['expiration']).timestamp())

            # For the purposes of this proposal, all cryptographic algorithms
            # are assumed to produce and consume byte strings, even if
            # internally they operate on some other data type like 64-bit
            # words. This is conventionally little endian order for Blake2b,
            # which contrasts with Tor's typical use of big endian.
            #
            # -- https://spec.torproject.org/hspow-spec/v1-equix.html
            effort = int.from_bytes(base64.b64decode(pow_data_v1['seed']), 'little')

            self.log('PoW v1 set with effort {}, expiration {} and seed {}'.format(
                pow_data_v1['effort'],
                pow_data_v1['expiration'],
                pow_data_v1['seed'],
                ))

            self.set_metric('onion_service_pow_v1_seed',               effort,                     labels)
            self.set_metric('onion_service_pow_v1_effort',             int(pow_data_v1['effort']), labels)
            self.set_metric('onion_service_pow_v1_expiration_seconds', int(expiration),            labels)

    def get_descriptor(self, endpoint, config, attempt = 1):
        """
        Get Onion Service descriptor from a given endpoint

        :type  endpoint: str
        :param endpoint: The endpoint name from the 'endpoints' instance config.

        :type  config: dict
        :param config: Endpoint configuration

        :rtype: stem.descriptor.hidden_service.InnerLayer or False
        :return: The Onion Service descriptor inner layer on success.
                 False on error.
        """

        self.log('Trying to get descriptor for {} (attempt {})...'.format(config['address'], attempt))

        pubkey    = self.get_pubkey_from_address(config['address'])
        init_time = self.now()
        timeout   = self.get_config('descriptor_timeout')
        reachable = 1

        # Metrics labels
        labels = {
                'name'   : endpoint,
                'address': config['address'],
                }

        # Get the descriptor
        try:
            # Increment the total number of descriptor fetch attempts
            self.inc_metric('onion_service_descriptor_fetch_requests_total', 1, labels)

            # Try to get the descriptor
            descriptor = self.controller.get_hidden_service_descriptor(pubkey, timeout=timeout)

        except (stem.DescriptorUnavailable, stem.Timeout, stem.ControllerError, ValueError) as e:
            reachable = 0
            inner     = False
            retries   = self.get_config('descriptor_max_retries')

            # Try again until max retries is reached
            if attempt <= retries:
                return self.get_descriptor(endpoint, config, attempt + 1)

        else:
            # Calculate the elapsed time
            elapsed = self.elapsed(init_time, True, "descriptor fetch")

            self.set_metric('onion_service_descriptor_latency_seconds',
                            elapsed, labels)

            # Update the HSDir latency metric
            if self.get_pubkey_from_address(config['address']) in self.hsdirs:
                # Register HSDir latency
                [ hsdir_id, hsdir_name ] = str(
                        self.hsdirs[self.get_pubkey_from_address(
                            config['address'])]).split('~')

                #self.log('HSDir ID: {}, HSDir name: {}'.format(hsdir_id, hsdir_name))
                self.set_metric('hsdir_latency_seconds',
                                elapsed, {
                                    'name': hsdir_name,
                                    'id'  : hsdir_id,
                                    })

            # Debuging the outer layer
            self.log("Outer wrapper descriptor layer contents (decrypted):\n" + str(descriptor), 'debug')

            self.set_metric('onion_service_descriptor_outer_wrapper_size_bytes',
                    len(str(descriptor).encode('utf-8')), labels)

            # Ensure it's converted to the v3 format
            #
            # See https://github.com/torproject/stem/issues/96
            #     https://stem.torproject.org/api/control.html#stem.control.Controller.get_hidden_service_descriptor
            #     https://gitlab.torproject.org/legacy/trac/-/issues/25417
            from stem.descriptor.hidden_service import HiddenServiceDescriptorV3
            descriptor = HiddenServiceDescriptorV3.from_str(str(descriptor))

            # Decrypt the inner layer
            inner = descriptor.decrypt(pubkey)

            self.set_metric('onion_service_descriptor_second_layer_size_bytes',
                    len(str(inner._raw_contents).encode('utf-8')), labels)

            if descriptor.lifetime:
                self.log("Descriptor lifetime: " + str(descriptor.lifetime))
                self.set_metric('onion_service_descriptor_lifetime_seconds',
                                descriptor.lifetime * 60, labels)

            if descriptor.revision_counter:
                self.log("Descriptor revision counter: " + str(descriptor.revision_counter))
                self.set_metric('onion_service_descriptor_revision_counter',
                                descriptor.revision_counter, labels)

            self.log("Single service mode is set to " + str(inner.is_single_service))
            self.set_metric('onion_service_is_single', inner.is_single_service, labels)

            # Debuging the inner layer
            self.log("Second layer of encryption descriptor contents (decrypted):\n" + inner._raw_contents, 'debug')

            # Get introduction points
            # See https://stem.torproject.org/api/descriptor/hidden_service.html#stem.descriptor.hidden_service.IntroductionPointV3
            #for introduction_point in inner.introduction_points:
            #    self.log(introduction_point.link_specifiers, 'debug')

            if 'introduction_points' in dir(inner):
                self.log("Number of introduction points: " + str(len(inner.introduction_points)))
                self.set_metric('onion_service_introduction_points_number',
                                len(inner.introduction_points), labels)

            # Parse PoW parameters
            self.parse_pow_params(inner._raw_contents, labels)

        finally:
            if inner is False:
                self.inc_metric('onion_service_descriptor_fetch_error_total', 1, labels)
            #else:
            #    # Increment the total number of sucessful descriptor fetch attempts
            #    self.inc_metric('onion_service_descriptor_fetch_success_total', 1, labels)

            labels['reachable'] = reachable

            # Register the number of fetch attempts in the current probing round
            self.set_metric('onion_service_descriptor_fetch_attempts',
                            attempt, labels)

            # Return the inner layer or False
            return inner

    def hsdesc_event(
            self,
            event,
            ):
        """
        Process HS_DESC events.

        Sets the onion_service_descriptor_reachable metric.

        See https://spec.torproject.org/control-spec/replies.html#HS_DESC
            https://spec.torproject.org/control-spec/replies.html#HS_DESC_CONTENT

        :type  event : stem.response.events.HSDescEvent
        :param stream: HS_DESC event
        """

        if event.action not in [ 'RECEIVED', 'FAILED' ]:
            return

        # Get the endpoint configuration
        (name, endpoint) = self.get_endpoint_by_pubkey(event.address)

        # Metrics labels
        labels = {
                'name'   : name,
                'address': event.address + '.onion',
                }

        if event.action == 'RECEIVED':
            reason = event.action

            self.set_metric('onion_service_descriptor_reachable', 1, labels)

        elif event.action == 'FAILED':
            # See control-spec.txt section "4.1.25. HiddenService descriptors"
            # FAILED action is split into it's reasons
            reason = event.reason

            self.set_metric('onion_service_descriptor_reachable', 0, labels)

        # Descriptor reachability
        self.log("Descriptor reachability: " + str(reason))

        # Log the HSDir
        self.log("HSDir used: " + str(event.directory))

        self.info_metric('onion_service_descriptor', {
            'hsdir': event.directory,
            'state': reason,
            },
            labels)

        # Initialize the HSDirs object if needed
        if 'hsdirs' not in dir(self):
            self.hsdirs = {}

        # Register the HSDir where the descriptor was fetched
        self.hsdirs[event.address] = str(event.directory).split('$')[1]

get_descriptor(endpoint, config, attempt=1)

Get Onion Service descriptor from a given endpoint

Parameters:

Name Type Description Default
endpoint

The endpoint name from the 'endpoints' instance config.

 required
config

Endpoint configuration

 required

Returns:

Type Description
stem.descriptor.hidden_service.InnerLayer | False

The Onion Service descriptor inner layer on success. False on error.
Source code in packages/onionprobe/descriptor.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def get_descriptor(self, endpoint, config, attempt = 1):
    """
    Get Onion Service descriptor from a given endpoint

    :type  endpoint: str
    :param endpoint: The endpoint name from the 'endpoints' instance config.

    :type  config: dict
    :param config: Endpoint configuration

    :rtype: stem.descriptor.hidden_service.InnerLayer or False
    :return: The Onion Service descriptor inner layer on success.
             False on error.
    """

    self.log('Trying to get descriptor for {} (attempt {})...'.format(config['address'], attempt))

    pubkey    = self.get_pubkey_from_address(config['address'])
    init_time = self.now()
    timeout   = self.get_config('descriptor_timeout')
    reachable = 1

    # Metrics labels
    labels = {
            'name'   : endpoint,
            'address': config['address'],
            }

    # Get the descriptor
    try:
        # Increment the total number of descriptor fetch attempts
        self.inc_metric('onion_service_descriptor_fetch_requests_total', 1, labels)

        # Try to get the descriptor
        descriptor = self.controller.get_hidden_service_descriptor(pubkey, timeout=timeout)

    except (stem.DescriptorUnavailable, stem.Timeout, stem.ControllerError, ValueError) as e:
        reachable = 0
        inner     = False
        retries   = self.get_config('descriptor_max_retries')

        # Try again until max retries is reached
        if attempt <= retries:
            return self.get_descriptor(endpoint, config, attempt + 1)

    else:
        # Calculate the elapsed time
        elapsed = self.elapsed(init_time, True, "descriptor fetch")

        self.set_metric('onion_service_descriptor_latency_seconds',
                        elapsed, labels)

        # Update the HSDir latency metric
        if self.get_pubkey_from_address(config['address']) in self.hsdirs:
            # Register HSDir latency
            [ hsdir_id, hsdir_name ] = str(
                    self.hsdirs[self.get_pubkey_from_address(
                        config['address'])]).split('~')

            #self.log('HSDir ID: {}, HSDir name: {}'.format(hsdir_id, hsdir_name))
            self.set_metric('hsdir_latency_seconds',
                            elapsed, {
                                'name': hsdir_name,
                                'id'  : hsdir_id,
                                })

        # Debuging the outer layer
        self.log("Outer wrapper descriptor layer contents (decrypted):\n" + str(descriptor), 'debug')

        self.set_metric('onion_service_descriptor_outer_wrapper_size_bytes',
                len(str(descriptor).encode('utf-8')), labels)

        # Ensure it's converted to the v3 format
        #
        # See https://github.com/torproject/stem/issues/96
        #     https://stem.torproject.org/api/control.html#stem.control.Controller.get_hidden_service_descriptor
        #     https://gitlab.torproject.org/legacy/trac/-/issues/25417
        from stem.descriptor.hidden_service import HiddenServiceDescriptorV3
        descriptor = HiddenServiceDescriptorV3.from_str(str(descriptor))

        # Decrypt the inner layer
        inner = descriptor.decrypt(pubkey)

        self.set_metric('onion_service_descriptor_second_layer_size_bytes',
                len(str(inner._raw_contents).encode('utf-8')), labels)

        if descriptor.lifetime:
            self.log("Descriptor lifetime: " + str(descriptor.lifetime))
            self.set_metric('onion_service_descriptor_lifetime_seconds',
                            descriptor.lifetime * 60, labels)

        if descriptor.revision_counter:
            self.log("Descriptor revision counter: " + str(descriptor.revision_counter))
            self.set_metric('onion_service_descriptor_revision_counter',
                            descriptor.revision_counter, labels)

        self.log("Single service mode is set to " + str(inner.is_single_service))
        self.set_metric('onion_service_is_single', inner.is_single_service, labels)

        # Debuging the inner layer
        self.log("Second layer of encryption descriptor contents (decrypted):\n" + inner._raw_contents, 'debug')

        # Get introduction points
        # See https://stem.torproject.org/api/descriptor/hidden_service.html#stem.descriptor.hidden_service.IntroductionPointV3
        #for introduction_point in inner.introduction_points:
        #    self.log(introduction_point.link_specifiers, 'debug')

        if 'introduction_points' in dir(inner):
            self.log("Number of introduction points: " + str(len(inner.introduction_points)))
            self.set_metric('onion_service_introduction_points_number',
                            len(inner.introduction_points), labels)

        # Parse PoW parameters
        self.parse_pow_params(inner._raw_contents, labels)

    finally:
        if inner is False:
            self.inc_metric('onion_service_descriptor_fetch_error_total', 1, labels)
        #else:
        #    # Increment the total number of sucessful descriptor fetch attempts
        #    self.inc_metric('onion_service_descriptor_fetch_success_total', 1, labels)

        labels['reachable'] = reachable

        # Register the number of fetch attempts in the current probing round
        self.set_metric('onion_service_descriptor_fetch_attempts',
                        attempt, labels)

        # Return the inner layer or False
        return inner

get_endpoint_by_pubkey(pubkey)

Get an endpoint configuration given an Onion Service pubkey.

Parameters:

Name Type Description Default
pubkey

Onion Service pubkey

 required

Returns:

Type Description
tuple | False

Endpoint name and configuration if a match is found. False otherwise.
Source code in packages/onionprobe/descriptor.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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_pubkey_from_address(address)

Extract .onion pubkey from the address

Leaves out the .onion domain suffix and any existing subdomains.

Parameters:

Name Type Description Default
address

Onion Service address

 required

Returns:

Type Description
str

Onion Service public key
Source code in packages/onionprobe/descriptor.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_pubkey_from_address(self, address):
    """
    Extract .onion pubkey from the address

    Leaves out the .onion domain suffix and any existing subdomains.

    :type  address: str
    :param address: Onion Service address

    :rtype: str
    :return: Onion Service public key
    """

    # Extract
    pubkey = address[0:-6].split('.')[-1]

    return pubkey

hsdesc_event(event)

Process HS_DESC events.

Sets the onion_service_descriptor_reachable metric.

See https://spec.torproject.org/control-spec/replies.html#HS_DESC https://spec.torproject.org/control-spec/replies.html#HS_DESC_CONTENT

Parameters:

Name Type Description Default
stream

HS_DESC event

 required
Source code in packages/onionprobe/descriptor.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def hsdesc_event(
        self,
        event,
        ):
    """
    Process HS_DESC events.

    Sets the onion_service_descriptor_reachable metric.

    See https://spec.torproject.org/control-spec/replies.html#HS_DESC
        https://spec.torproject.org/control-spec/replies.html#HS_DESC_CONTENT

    :type  event : stem.response.events.HSDescEvent
    :param stream: HS_DESC event
    """

    if event.action not in [ 'RECEIVED', 'FAILED' ]:
        return

    # Get the endpoint configuration
    (name, endpoint) = self.get_endpoint_by_pubkey(event.address)

    # Metrics labels
    labels = {
            'name'   : name,
            'address': event.address + '.onion',
            }

    if event.action == 'RECEIVED':
        reason = event.action

        self.set_metric('onion_service_descriptor_reachable', 1, labels)

    elif event.action == 'FAILED':
        # See control-spec.txt section "4.1.25. HiddenService descriptors"
        # FAILED action is split into it's reasons
        reason = event.reason

        self.set_metric('onion_service_descriptor_reachable', 0, labels)

    # Descriptor reachability
    self.log("Descriptor reachability: " + str(reason))

    # Log the HSDir
    self.log("HSDir used: " + str(event.directory))

    self.info_metric('onion_service_descriptor', {
        'hsdir': event.directory,
        'state': reason,
        },
        labels)

    # Initialize the HSDirs object if needed
    if 'hsdirs' not in dir(self):
        self.hsdirs = {}

    # Register the HSDir where the descriptor was fetched
    self.hsdirs[event.address] = str(event.directory).split('$')[1]

parse_pow_params(inner_text, labels)

Parse the Proof of Work (PoW) parameters from a descriptor.

Parameters:

Name Type Description Default
inner_text

The decrypted raw inner descriptor layer plaintext for the endpoint.

 required
labels

Metrics labels

 required

Returns:

Type Description
None

This method does not return any special value.
Source code in packages/onionprobe/descriptor.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def parse_pow_params(self, inner_text, labels):
    """
    Parse the Proof of Work (PoW) parameters from a descriptor.

    :type  inner_text: str
    :param inner_text: The decrypted raw inner descriptor layer plaintext for the endpoint.

    :type  labels: dict
    :param labels: Metrics labels

    :rtype:  None
    :return: This method does not return any special value.
    """

    pow_params    = re.compile(r"^pow-params .*$", re.MULTILINE)
    pow_params_v1 = re.compile(
            r"^pow-params v1 (?P<seed>[^ ]*) (?P<effort>[0-9]*) (?P<expiration>[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2})$",
            re.MULTILINE)

    pow_parsed    = pow_params.search(inner_text)
    pow_parsed_v1 = pow_params_v1.search(inner_text)

    if pow_parsed:
        self.log("Proof of Work (PoW) params found in the descriptor")
        self.set_metric('onion_service_pow_enabled', 1, labels)
    else:
        self.log("Proof of Work (PoW) params not found in the descriptor")
        self.set_metric('onion_service_pow_enabled', 0, labels)

    if pow_parsed_v1:
        pow_data_v1 = pow_parsed_v1.groupdict()
        expiration  = int(datetime.datetime.fromisoformat(pow_data_v1['expiration']).timestamp())

        # For the purposes of this proposal, all cryptographic algorithms
        # are assumed to produce and consume byte strings, even if
        # internally they operate on some other data type like 64-bit
        # words. This is conventionally little endian order for Blake2b,
        # which contrasts with Tor's typical use of big endian.
        #
        # -- https://spec.torproject.org/hspow-spec/v1-equix.html
        effort = int.from_bytes(base64.b64decode(pow_data_v1['seed']), 'little')

        self.log('PoW v1 set with effort {}, expiration {} and seed {}'.format(
            pow_data_v1['effort'],
            pow_data_v1['expiration'],
            pow_data_v1['seed'],
            ))

        self.set_metric('onion_service_pow_v1_seed',               effort,                     labels)
        self.set_metric('onion_service_pow_v1_effort',             int(pow_data_v1['effort']), labels)
        self.set_metric('onion_service_pow_v1_expiration_seconds', int(expiration),            labels)

http

OnionprobeHTTP

Onionprobe class with HTTP methods.

Source code in packages/onionprobe/http.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
class OnionprobeHTTP:
    """
    Onionprobe class with HTTP methods.
    """

    def build_url(self, config, path = None):
        """
        Build an Onion Service URL to be probed

        :type  config: dict
        :param config: Endpoint configuration

        :type  path: str
        :param path: The path to be chosen in the endpoint configuration.

        :rtype: str
        :return: The Onion Service URL for the given config and path

        """

        # Get the base URL
        url = config['address']

        # Set the protocol
        if 'protocol' in config:
            url = config['protocol'] + '://' + url

        # Set the port
        if 'port' in config:
            url += ':' + str(config['port'])

        # Set the path
        #if 'path' in config:
        #    url += config['path']
        if path is not None:
            url += path

        return url

    def query_http(self, endpoint, config, path, attempt = 1):
        """
        Fetches endpoint from URL

        Tries an HTTP connection to the URL and update metrics when needed.

        :type  endpoint: str
        :param endpoint: The endpoint name from the 'endpoints' instance config.

        :type  config: dict
        :param config: Endpoint configuration

        :type  path: dict
        :param path: A path dictionary from the endpoint configuration.

        :type  attempt: int
        :param attempt: The current attempt used to determine the maximum number of retries.

        :rtype: requests.Response or False
        :return: The query result on success.
                 False on error.
        """

        # Parameter checks
        if not isinstance(path, dict):
            self.log('Path parameter should be dictionary, {} given'.format(type(path)), 'error')

            return False

        # Setup query parameters
        url         = self.build_url(config, path['path'])
        result      = False
        exception   = None
        init_time   = self.now()
        tor_address = self.get_config('tor_address')
        socks_port  = self.get_config('socks_port')

        # Request everything via Tor, including DNS queries
        proxies = {
                'http' : 'socks5h://{}:{}'.format(tor_address, socks_port),
                'https': 'socks5h://{}:{}'.format(tor_address, socks_port),
                }

        # Metric labels
        labels = {
                'name'     : endpoint,
                'address'  : config['address'],
                'protocol' : config['protocol'],
                'port'     : config['port'],
                'path'     : path['path'],
                }

        timeout = (
                self.get_config('http_connect_timeout'),
                self.get_config('http_read_timeout'),
                )

        # Whether to verify TLS certificates
        if 'tls_verify' in config:
            tls_verify = config['tls_verify']
        else:
            tls_verify = self.config.get('tls_verify')

        # Untested certs get a default status value as well
        valid_cert = 1 if tls_verify else 2

        try:
            self.log('Trying to connect to {} (attempt {})...'.format(url, attempt))
            self.inc_metric('onion_service_fetch_requests_total', 1, labels)

            # Fetch results and calculate the elapsed time
            result  = requests.get(url, proxies=proxies, timeout=timeout, verify=tls_verify)
            elapsed = self.elapsed(init_time, True, "HTTP fetch")

            # Update metrics
            self.set_metric('onion_service_latency_seconds', elapsed, labels)

        except requests.exceptions.TooManyRedirects as e:
            result    = False
            exception = 'too_many_redirects'

            self.log(e, 'error')

        except requests.exceptions.SSLError as e:
            result     = False
            exception  = 'certificate_error'
            valid_cert = 0

            self.log(e, 'error')

        # Requests that produced this error are safe to retry, but we are not
        # doing that right now
        except requests.exceptions.ConnectionTimeout as e:
            result    = False
            exception = 'connection_timeout'

            self.log(e, 'error')

        except requests.exceptions.ReadTimeout as e:
            result    = False
            exception = 'connection_read_timeout'

            self.log(e, 'error')

        except requests.exceptions.Timeout as e:
            result    = False
            exception = 'timeout'

            self.log(e, 'error')

        except requests.exceptions.HTTPError as e:
            result    = False
            exception = 'http_error'

            self.log(e, 'error')

        except requests.exceptions.ConnectionError as e:
            result    = False
            exception = 'connection_error'

            self.log(e, 'error')

        except requests.exceptions.RequestException as e:
            result    = False
            exception = 'request_exception'

            self.log(e, 'error')

        except Exception as e:
            result    = False
            exception = 'generic_error'

            self.log(e, 'error')

        else:
            self.log('Status code is {}'.format(result.status_code))

            # Register status code in the metrics
            self.set_metric('onion_service_status_code', result.status_code, labels)

            # Check for expected status codes
            if 'allowed_statuses' in path:
                if result.status_code not in path['allowed_statuses']:
                    result          = False
                    expected_status = 1
                    expected_clause = 'none'
                    expected_level  = 'error'

                else:
                    expected_status = 0
                    expected_clause = 'one'
                    expected_level  = 'info'

                self.log('Status code match {} of the expected {}'.format(
                    expected_clause, repr(path['allowed_statuses'])),
                    expected_level
                    )

                self.set_metric('onion_service_unexpected_status_code', expected_status, labels)

        finally:
            reachable = 0 if result is False else 1

            if result is False:
                retries = self.get_config('http_connect_max_retries')

                # Try again until max retries is reached
                if attempt <= retries:
                    return self.query_http(endpoint, config, path, attempt + 1)

            # Register reachability on metrics
            self.set_metric('onion_service_reachable', reachable, labels)

            if config['protocol'] == 'https':
                self.set_metric('onion_service_valid_certificate', valid_cert, labels)

            if exception is not None:
                # Count exceptions
                self.inc_metric('onion_service_' + exception + '_total', 1, labels)

                # Count errors
                self.inc_metric('onion_service_fetch_error_total', 1, labels)

            #if expected_status == 1:
            #    # Count unexpected statuses
            #    self.set_metric('onion_service_unexpected_status_code_total', 1, labels)

            #else:
            #    # Increment the total number of successful fetches
            #    self.inc_metric('onion_service_fetch_success_total', 1, labels)

            # Register the number of attempts on metrics
            labels['reachable'] = reachable
            self.set_metric('onion_service_connection_attempts', attempt, labels)

            return result

build_url(config, path=None)

Build an Onion Service URL to be probed

Parameters:

Name Type Description Default
config

Endpoint configuration

 required
path

The path to be chosen in the endpoint configuration.

 None

Returns:

Type Description
str

The Onion Service URL for the given config and path
Source code in packages/onionprobe/http.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def build_url(self, config, path = None):
    """
    Build an Onion Service URL to be probed

    :type  config: dict
    :param config: Endpoint configuration

    :type  path: str
    :param path: The path to be chosen in the endpoint configuration.

    :rtype: str
    :return: The Onion Service URL for the given config and path

    """

    # Get the base URL
    url = config['address']

    # Set the protocol
    if 'protocol' in config:
        url = config['protocol'] + '://' + url

    # Set the port
    if 'port' in config:
        url += ':' + str(config['port'])

    # Set the path
    #if 'path' in config:
    #    url += config['path']
    if path is not None:
        url += path

    return url

query_http(endpoint, config, path, attempt=1)

Fetches endpoint from URL

Tries an HTTP connection to the URL and update metrics when needed.

Parameters:

Name Type Description Default
endpoint

The endpoint name from the 'endpoints' instance config.

 required
config

Endpoint configuration

 required
path

A path dictionary from the endpoint configuration.

 required
attempt

The current attempt used to determine the maximum number of retries.

 1

Returns:

Type Description
requests.Response | False

The query result on success. False on error.
Source code in packages/onionprobe/http.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def query_http(self, endpoint, config, path, attempt = 1):
    """
    Fetches endpoint from URL

    Tries an HTTP connection to the URL and update metrics when needed.

    :type  endpoint: str
    :param endpoint: The endpoint name from the 'endpoints' instance config.

    :type  config: dict
    :param config: Endpoint configuration

    :type  path: dict
    :param path: A path dictionary from the endpoint configuration.

    :type  attempt: int
    :param attempt: The current attempt used to determine the maximum number of retries.

    :rtype: requests.Response or False
    :return: The query result on success.
             False on error.
    """

    # Parameter checks
    if not isinstance(path, dict):
        self.log('Path parameter should be dictionary, {} given'.format(type(path)), 'error')

        return False

    # Setup query parameters
    url         = self.build_url(config, path['path'])
    result      = False
    exception   = None
    init_time   = self.now()
    tor_address = self.get_config('tor_address')
    socks_port  = self.get_config('socks_port')

    # Request everything via Tor, including DNS queries
    proxies = {
            'http' : 'socks5h://{}:{}'.format(tor_address, socks_port),
            'https': 'socks5h://{}:{}'.format(tor_address, socks_port),
            }

    # Metric labels
    labels = {
            'name'     : endpoint,
            'address'  : config['address'],
            'protocol' : config['protocol'],
            'port'     : config['port'],
            'path'     : path['path'],
            }

    timeout = (
            self.get_config('http_connect_timeout'),
            self.get_config('http_read_timeout'),
            )

    # Whether to verify TLS certificates
    if 'tls_verify' in config:
        tls_verify = config['tls_verify']
    else:
        tls_verify = self.config.get('tls_verify')

    # Untested certs get a default status value as well
    valid_cert = 1 if tls_verify else 2

    try:
        self.log('Trying to connect to {} (attempt {})...'.format(url, attempt))
        self.inc_metric('onion_service_fetch_requests_total', 1, labels)

        # Fetch results and calculate the elapsed time
        result  = requests.get(url, proxies=proxies, timeout=timeout, verify=tls_verify)
        elapsed = self.elapsed(init_time, True, "HTTP fetch")

        # Update metrics
        self.set_metric('onion_service_latency_seconds', elapsed, labels)

    except requests.exceptions.TooManyRedirects as e:
        result    = False
        exception = 'too_many_redirects'

        self.log(e, 'error')

    except requests.exceptions.SSLError as e:
        result     = False
        exception  = 'certificate_error'
        valid_cert = 0

        self.log(e, 'error')

    # Requests that produced this error are safe to retry, but we are not
    # doing that right now
    except requests.exceptions.ConnectionTimeout as e:
        result    = False
        exception = 'connection_timeout'

        self.log(e, 'error')

    except requests.exceptions.ReadTimeout as e:
        result    = False
        exception = 'connection_read_timeout'

        self.log(e, 'error')

    except requests.exceptions.Timeout as e:
        result    = False
        exception = 'timeout'

        self.log(e, 'error')

    except requests.exceptions.HTTPError as e:
        result    = False
        exception = 'http_error'

        self.log(e, 'error')

    except requests.exceptions.ConnectionError as e:
        result    = False
        exception = 'connection_error'

        self.log(e, 'error')

    except requests.exceptions.RequestException as e:
        result    = False
        exception = 'request_exception'

        self.log(e, 'error')

    except Exception as e:
        result    = False
        exception = 'generic_error'

        self.log(e, 'error')

    else:
        self.log('Status code is {}'.format(result.status_code))

        # Register status code in the metrics
        self.set_metric('onion_service_status_code', result.status_code, labels)

        # Check for expected status codes
        if 'allowed_statuses' in path:
            if result.status_code not in path['allowed_statuses']:
                result          = False
                expected_status = 1
                expected_clause = 'none'
                expected_level  = 'error'

            else:
                expected_status = 0
                expected_clause = 'one'
                expected_level  = 'info'

            self.log('Status code match {} of the expected {}'.format(
                expected_clause, repr(path['allowed_statuses'])),
                expected_level
                )

            self.set_metric('onion_service_unexpected_status_code', expected_status, labels)

    finally:
        reachable = 0 if result is False else 1

        if result is False:
            retries = self.get_config('http_connect_max_retries')

            # Try again until max retries is reached
            if attempt <= retries:
                return self.query_http(endpoint, config, path, attempt + 1)

        # Register reachability on metrics
        self.set_metric('onion_service_reachable', reachable, labels)

        if config['protocol'] == 'https':
            self.set_metric('onion_service_valid_certificate', valid_cert, labels)

        if exception is not None:
            # Count exceptions
            self.inc_metric('onion_service_' + exception + '_total', 1, labels)

            # Count errors
            self.inc_metric('onion_service_fetch_error_total', 1, labels)

        #if expected_status == 1:
        #    # Count unexpected statuses
        #    self.set_metric('onion_service_unexpected_status_code_total', 1, labels)

        #else:
        #    # Increment the total number of successful fetches
        #    self.inc_metric('onion_service_fetch_success_total', 1, labels)

        # Register the number of attempts on metrics
        labels['reachable'] = reachable
        self.set_metric('onion_service_connection_attempts', attempt, labels)

        return result

init

OnionprobeInit

Source code in packages/onionprobe/init.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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

__init__(args)

Onionprobe class constructor.

Setup instance configuration.

Handles command-line parameters.

Parameters:

Name Type Description Default
args

Instance arguments.

 required
Source code in packages/onionprobe/init.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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

initialize()

Onionprobe initialization procedures

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

Returns:

Type Description
bol

True if initialization is successful, False on error
Source code in packages/onionprobe/init.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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

logger

OnionprobeLogger

Onionprobe class with logging methods.

Source code in packages/onionprobe/logger.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class OnionprobeLogger:
    """
    Onionprobe class with logging methods.
    """

    def initialize_logging(self):
        """
        Initialize Onionprobe's logging subsystem

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

        log_level = self.get_config('log_level').upper()

        if log_level in dir(logging):
            level = getattr(logging, log_level)

            logging.basicConfig(level=level, format='%(asctime)s %(levelname)s: %(message)s')

            # See https://stem.torproject.org/api/util/log.html
            stem_logger = stem.util.log.get_logger()

            stem_logger.setLevel(level)

        else:
            logging.error("Invalid log level %s" % (log_level))

            return False

        self.log('Starting Onionprobe version %s...' % (onionprobe_version))

        return True

    def log(self, message, level='info'):
        """
        Helper log function

        Appends a message into the logging subsystem.

        :type  message: str
        :param message: The message to be logged.

        :type  level: str
        :param level: The log level. Defaults to 'info'.
                      For the available log levels, check
                      https://docs.python.org/3/howto/logging.html#logging-levels
        """

        # Just a wrapper for the logging() function
        getattr(logging, level)(message)

initialize_logging()

Initialize Onionprobe's logging subsystem

Returns:

Type Description
bol

True if initialization is successful, False on error
Source code in packages/onionprobe/logger.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def initialize_logging(self):
    """
    Initialize Onionprobe's logging subsystem

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

    log_level = self.get_config('log_level').upper()

    if log_level in dir(logging):
        level = getattr(logging, log_level)

        logging.basicConfig(level=level, format='%(asctime)s %(levelname)s: %(message)s')

        # See https://stem.torproject.org/api/util/log.html
        stem_logger = stem.util.log.get_logger()

        stem_logger.setLevel(level)

    else:
        logging.error("Invalid log level %s" % (log_level))

        return False

    self.log('Starting Onionprobe version %s...' % (onionprobe_version))

    return True

log(message, level='info')

Helper log function

Appends a message into the logging subsystem.

Parameters:

Name Type Description Default
message

The message to be logged.

 required
level

The log level. Defaults to 'info'. For the available log levels, check https://docs.python.org/3/howto/logging.html#logging-levels

 'info'
Source code in packages/onionprobe/logger.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def log(self, message, level='info'):
    """
    Helper log function

    Appends a message into the logging subsystem.

    :type  message: str
    :param message: The message to be logged.

    :type  level: str
    :param level: The log level. Defaults to 'info'.
                  For the available log levels, check
                  https://docs.python.org/3/howto/logging.html#logging-levels
    """

    # Just a wrapper for the logging() function
    getattr(logging, level)(message)

main

OnionprobeMain

Onionprobe class with main application logic.

Source code in packages/onionprobe/main.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class OnionprobeMain:
    """
    Onionprobe class with main application logic.
    """

    def run(self):
        """
        Main application loop

        Checks if should be run indefinitely.
        Then dispatch to a round of probes.

        If runs continuously, waits before starting the next round.

        If not, just returns.

        :rtype:  bol
        :return: True on success, false if at least one of the probes fails.
        """

        status = True

        # Check if should loop
        if self.get_config('loop'):
            iteration = 1
            rounds    = self.get_config('rounds')

            while True:
                self.log('Starting round %s, probing all defined endpoints...' % (iteration))

                # Call for a round
                result = self.round()

                if result is False:
                    status = False

                # Check rounds
                if rounds > 0 and iteration >= rounds:
                    self.log('Stopping after %s rounds' % (iteration))

                    break

                self.log('Round %s completed.' % (iteration))

                # Then wait
                self.wait(self.get_config('sleep'))

                # Update iterations counter
                iteration += 1

        else:
            # Single pass, only one round
            status = self.round()

        return status

    def round(self):
        """
        Process a round of probes

        Each round is composed of the entire set of the endpoints
        which is optionally shuffled.

        Each endpoint is then probed.

        :rtype:  bol
        :return: True on success, false if at least one of the probes fails.
        """

        # Shuffle the deck
        endpoints = sorted(self.get_config('endpoints'))

        # Hold general probe status
        status = True

        if self.get_config('shuffle'):
            # Reinitializes the random number generator to avoid predictable
            # results if running countinuously for long periods.
            random.seed()

            endpoints = random.sample(endpoints, k=len(endpoints))

        # Probe each endpoint
        for key, endpoint in enumerate(endpoints):
            self.metrics['onionprobe_state'].state('probing')

            result = self.probe(endpoint)

            if result is None or result is False:
                status = False
            else:
                for item in result:
                    if result[item] == False:
                        status = False
                        break

            # Wait if not last endpoint
            if key != len(endpoints) - 1:
                self.wait(self.get_config('interval'))

        return status

round()

Process a round of probes

Each round is composed of the entire set of the endpoints which is optionally shuffled.

Each endpoint is then probed.

Returns:

Type Description
bol

True on success, false if at least one of the probes fails.
Source code in packages/onionprobe/main.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def round(self):
    """
    Process a round of probes

    Each round is composed of the entire set of the endpoints
    which is optionally shuffled.

    Each endpoint is then probed.

    :rtype:  bol
    :return: True on success, false if at least one of the probes fails.
    """

    # Shuffle the deck
    endpoints = sorted(self.get_config('endpoints'))

    # Hold general probe status
    status = True

    if self.get_config('shuffle'):
        # Reinitializes the random number generator to avoid predictable
        # results if running countinuously for long periods.
        random.seed()

        endpoints = random.sample(endpoints, k=len(endpoints))

    # Probe each endpoint
    for key, endpoint in enumerate(endpoints):
        self.metrics['onionprobe_state'].state('probing')

        result = self.probe(endpoint)

        if result is None or result is False:
            status = False
        else:
            for item in result:
                if result[item] == False:
                    status = False
                    break

        # Wait if not last endpoint
        if key != len(endpoints) - 1:
            self.wait(self.get_config('interval'))

    return status

run()

Main application loop

Checks if should be run indefinitely. Then dispatch to a round of probes.

If runs continuously, waits before starting the next round.

If not, just returns.

Returns:

Type Description
bol

True on success, false if at least one of the probes fails.
Source code in packages/onionprobe/main.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def run(self):
    """
    Main application loop

    Checks if should be run indefinitely.
    Then dispatch to a round of probes.

    If runs continuously, waits before starting the next round.

    If not, just returns.

    :rtype:  bol
    :return: True on success, false if at least one of the probes fails.
    """

    status = True

    # Check if should loop
    if self.get_config('loop'):
        iteration = 1
        rounds    = self.get_config('rounds')

        while True:
            self.log('Starting round %s, probing all defined endpoints...' % (iteration))

            # Call for a round
            result = self.round()

            if result is False:
                status = False

            # Check rounds
            if rounds > 0 and iteration >= rounds:
                self.log('Stopping after %s rounds' % (iteration))

                break

            self.log('Round %s completed.' % (iteration))

            # Then wait
            self.wait(self.get_config('sleep'))

            # Update iterations counter
            iteration += 1

    else:
        # Single pass, only one round
        status = self.round()

    return status

metrics

OnionprobeMetrics

Onionprobe class with metrics methods.

Source code in packages/onionprobe/metrics.py 
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
class OnionprobeMetrics:
    """
    Onionprobe class with metrics methods.
    """

    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)

    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')

    def set_metric(self, metric, value, labels = {}):
        """
        Set a metric.

        :type  metric: str
        :param metric: Metric name

        :type  value: int
        :param value: Metric value

        :type  labels: dict
        :param labels: Metric labels dictionary.
                       Defaults to an empty dictionary.
        """

        if metric in self.metrics:
            self.metrics[metric].labels(**labels).set(value)

    def inc_metric(self, metric, value = 1, labels = {}):
        """
        Increment a metric.

        :type  metric: str
        :param metric: Metric name

        :type  value: int
        :param value: Increment value. Defaults to 1.

        :type  labels: dict
        :param labels: Metric labels dictionary.
                       Defaults to an empty dictionary.
        """

        if metric in self.metrics:
            self.metrics[metric].labels(**labels).inc(value)

    def state_metric(self, metric, value, labels = {}):
        """
        Set a metric state.

        :type  metric: str
        :param metric: Metric name

        :type  value: Object
        :param value: Increment value.

        :type  labels: dict
        :param labels: Metric labels dictionary.
                       Defaults to an empty dictionary.
        """

        if metric in self.metrics:
            self.metrics[metric].labels(**labels).state(value)

    def info_metric(self, metric, value, labels = {}):
        """
        Set an info metric.

        :type  metric: str
        :param metric: Metric name

        :type  value: dict
        :param value: Increment value.

        :type  labels: dict
        :param labels: Metric labels dictionary.
                       Defaults to an empty dictionary.
        """

        if metric in self.metrics:
            self.metrics[metric].labels(**labels).info(value)

inc_metric(metric, value=1, labels={})

Increment a metric.

Parameters:

Name Type Description Default
metric

Metric name

 required
value

Increment value. Defaults to 1.

 1
labels

Metric labels dictionary. Defaults to an empty dictionary.

 {}
Source code in packages/onionprobe/metrics.py 
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
def inc_metric(self, metric, value = 1, labels = {}):
    """
    Increment a metric.

    :type  metric: str
    :param metric: Metric name

    :type  value: int
    :param value: Increment value. Defaults to 1.

    :type  labels: dict
    :param labels: Metric labels dictionary.
                   Defaults to an empty dictionary.
    """

    if metric in self.metrics:
        self.metrics[metric].labels(**labels).inc(value)

info_metric(metric, value, labels={})

Set an info metric.

Parameters:

Name Type Description Default
metric

Metric name

 required
value

Increment value.

 required
labels

Metric labels dictionary. Defaults to an empty dictionary.

 {}
Source code in packages/onionprobe/metrics.py 
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
def info_metric(self, metric, value, labels = {}):
    """
    Set an info metric.

    :type  metric: str
    :param metric: Metric name

    :type  value: dict
    :param value: Increment value.

    :type  labels: dict
    :param labels: Metric labels dictionary.
                   Defaults to an empty dictionary.
    """

    if metric in self.metrics:
        self.metrics[metric].labels(**labels).info(value)

initialize_metrics()

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.

Source code in packages/onionprobe/metrics.py 
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
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')

initialize_prometheus_exporter()

Initialize the Prometheus Exporter

Source code in packages/onionprobe/metrics.py 
496
497
498
499
500
501
502
503
504
505
506
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)

set_metric(metric, value, labels={})

Set a metric.

Parameters:

Name Type Description Default
metric

Metric name

 required
value

Metric value

 required
labels

Metric labels dictionary. Defaults to an empty dictionary.

 {}
Source code in packages/onionprobe/metrics.py 
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
def set_metric(self, metric, value, labels = {}):
    """
    Set a metric.

    :type  metric: str
    :param metric: Metric name

    :type  value: int
    :param value: Metric value

    :type  labels: dict
    :param labels: Metric labels dictionary.
                   Defaults to an empty dictionary.
    """

    if metric in self.metrics:
        self.metrics[metric].labels(**labels).set(value)

state_metric(metric, value, labels={})

Set a metric state.

Parameters:

Name Type Description Default
metric

Metric name

 required
value

Increment value.

 required
labels

Metric labels dictionary. Defaults to an empty dictionary.

 {}
Source code in packages/onionprobe/metrics.py 
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
def state_metric(self, metric, value, labels = {}):
    """
    Set a metric state.

    :type  metric: str
    :param metric: Metric name

    :type  value: Object
    :param value: Increment value.

    :type  labels: dict
    :param labels: Metric labels dictionary.
                   Defaults to an empty dictionary.
    """

    if metric in self.metrics:
        self.metrics[metric].labels(**labels).state(value)

prober

OnionprobeProber

Onionprobe class with probing methods.

Source code in packages/onionprobe/prober.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class OnionprobeProber:
    """
    Onionprobe class with probing methods.
    """

    def probe(self, endpoint):
        """
        Probe an unique endpoint

        Checks for a valid and published Onion Service descriptor for the endpoint.

        Then probes each path configured for the endpoint, storing the results in
        a dictionary.

        Ensure that each probe starts with a cleared Stem Controller cache.

        :type  endpoint: str
        :param endpoint: The endpoint name from the 'endpoints' instance config.

        :rtype: dict or False
        :return: A dictionary of results for each path configured for the endpoint.
                 False in case of Onion Service descriptor error.
        """

        self.log("Processing {}...".format(endpoint))

        endpoints = self.get_config('endpoints')
        config    = endpoints[endpoint]

        # Check if the addres is valid
        from stem.util.tor_tools import is_valid_hidden_service_address

        if 'address' not in config:
            self.log('No address set for {}'.format(endpoint), 'error')

            return False

        elif is_valid_hidden_service_address(
                self.get_pubkey_from_address(config['address']), 3) is False:
            self.log('Invalid onion service address set for {}: {}'.format(
                endpoint, config['address']), 'error')

            return False

        # Register test metadata
        self.info_metric('onion_service_probe_status', {
            'last_tested_at_posix_timestamp': str(self.timestamp()),
            },
            {
            'name'   : endpoint,
            'address': config['address'],
            })

        # Ensure we always begin with a cleared cache
        # This allows to discover issues with published descriptors
        self.controller.clear_cache()

        # Ensure we use a new circuit every time
        # Needs to close all other circuits?
        # Needs to setup a 'controler' circuit?
        # Replaced by event listener at the initialize() method
        #circuit = self.controller.new_circuit()

        # Get Onion Service descriptor
        descriptor = self.get_descriptor(endpoint, config)

        if descriptor is False:
            self.log('Error getting the descriptor', 'error')

            return False

        # Ensure at least a single path
        if 'paths' not in config:
            config['paths'] = [
                        {
                            'path'            : '/',
                            'pattern'         : None,
                            'allowed_statuses': [ 200 ],
                        },
                    ]

        results = {}

        # Query each path
        for path in config['paths']:
            result = self.query_http(endpoint, config, path)

            if result is not False:
                # Check for a match
                if 'pattern' in path and path['pattern'] is not None:
                    import re
                    pattern = re.compile(path['pattern'])
                    match   = pattern.search(result.text)

                    self.log('Looking for pattern {}...'.format(path['pattern']))

                    if match is not None:
                        self.log('Match found: "%s"' % (path['pattern']))

                        matched               = 1
                        results[path['path']] = result
                    else:
                        self.log('Match not found: "%s"' % (path['pattern']))

                        matched               = 0
                        results[path['path']] = False

                    # Update metrics
                    self.set_metric('onion_service_pattern_matched',
                                    matched, {
                                        'name'     : endpoint,
                                        'address'  : config['address'],
                                        'protocol' : config['protocol'],
                                        'port'     : config['port'],
                                        'path'     : path,
                                        'pattern'  : path['pattern'],
                                    })

                else:
                    results[path['path']] = result

            else:
                self.log('Error querying {}'.format(config['address']), 'error')

                results[path['path']] = False

        # Get certificate information
        if config['protocol'] == 'https':
            if ('test_tls_connection' in config and config['test_tls_connection']) or \
                    self.get_config('test_tls_connection'):
                cert = self.query_tls(endpoint, config)

        return results

probe(endpoint)

Probe an unique endpoint

Checks for a valid and published Onion Service descriptor for the endpoint.

Then probes each path configured for the endpoint, storing the results in a dictionary.

Ensure that each probe starts with a cleared Stem Controller cache.

Parameters:

Name Type Description Default
endpoint

The endpoint name from the 'endpoints' instance config.

 required

Returns:

Type Description
dict | False

A dictionary of results for each path configured for the endpoint. False in case of Onion Service descriptor error.
Source code in packages/onionprobe/prober.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def probe(self, endpoint):
    """
    Probe an unique endpoint

    Checks for a valid and published Onion Service descriptor for the endpoint.

    Then probes each path configured for the endpoint, storing the results in
    a dictionary.

    Ensure that each probe starts with a cleared Stem Controller cache.

    :type  endpoint: str
    :param endpoint: The endpoint name from the 'endpoints' instance config.

    :rtype: dict or False
    :return: A dictionary of results for each path configured for the endpoint.
             False in case of Onion Service descriptor error.
    """

    self.log("Processing {}...".format(endpoint))

    endpoints = self.get_config('endpoints')
    config    = endpoints[endpoint]

    # Check if the addres is valid
    from stem.util.tor_tools import is_valid_hidden_service_address

    if 'address' not in config:
        self.log('No address set for {}'.format(endpoint), 'error')

        return False

    elif is_valid_hidden_service_address(
            self.get_pubkey_from_address(config['address']), 3) is False:
        self.log('Invalid onion service address set for {}: {}'.format(
            endpoint, config['address']), 'error')

        return False

    # Register test metadata
    self.info_metric('onion_service_probe_status', {
        'last_tested_at_posix_timestamp': str(self.timestamp()),
        },
        {
        'name'   : endpoint,
        'address': config['address'],
        })

    # Ensure we always begin with a cleared cache
    # This allows to discover issues with published descriptors
    self.controller.clear_cache()

    # Ensure we use a new circuit every time
    # Needs to close all other circuits?
    # Needs to setup a 'controler' circuit?
    # Replaced by event listener at the initialize() method
    #circuit = self.controller.new_circuit()

    # Get Onion Service descriptor
    descriptor = self.get_descriptor(endpoint, config)

    if descriptor is False:
        self.log('Error getting the descriptor', 'error')

        return False

    # Ensure at least a single path
    if 'paths' not in config:
        config['paths'] = [
                    {
                        'path'            : '/',
                        'pattern'         : None,
                        'allowed_statuses': [ 200 ],
                    },
                ]

    results = {}

    # Query each path
    for path in config['paths']:
        result = self.query_http(endpoint, config, path)

        if result is not False:
            # Check for a match
            if 'pattern' in path and path['pattern'] is not None:
                import re
                pattern = re.compile(path['pattern'])
                match   = pattern.search(result.text)

                self.log('Looking for pattern {}...'.format(path['pattern']))

                if match is not None:
                    self.log('Match found: "%s"' % (path['pattern']))

                    matched               = 1
                    results[path['path']] = result
                else:
                    self.log('Match not found: "%s"' % (path['pattern']))

                    matched               = 0
                    results[path['path']] = False

                # Update metrics
                self.set_metric('onion_service_pattern_matched',
                                matched, {
                                    'name'     : endpoint,
                                    'address'  : config['address'],
                                    'protocol' : config['protocol'],
                                    'port'     : config['port'],
                                    'path'     : path,
                                    'pattern'  : path['pattern'],
                                })

            else:
                results[path['path']] = result

        else:
            self.log('Error querying {}'.format(config['address']), 'error')

            results[path['path']] = False

    # Get certificate information
    if config['protocol'] == 'https':
        if ('test_tls_connection' in config and config['test_tls_connection']) or \
                self.get_config('test_tls_connection'):
            cert = self.query_tls(endpoint, config)

    return results

teardown

OnionprobeTeardown

Onionprobe class with methods related to... stop running!

Source code in packages/onionprobe/teardown.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class OnionprobeTeardown:
    """
    Onionprobe class with methods related to... stop running!
    """

    def close(self):
        """
        Onionprobe teardown procedure.

        Change the internal metrics state to running.

        Stops the built-in Tor daemon.
        """

        if 'metrics' in dir(self):
            self.metrics['onionprobe_state'].state('stopping')

        if 'controller' in dir(self):
            self.controller.close()

        # Terminate built-in Tor
        if 'tor' in dir(self):
            self.tor.kill()

close()

Onionprobe teardown procedure.

Change the internal metrics state to running.

Stops the built-in Tor daemon.

Source code in packages/onionprobe/teardown.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def close(self):
    """
    Onionprobe teardown procedure.

    Change the internal metrics state to running.

    Stops the built-in Tor daemon.
    """

    if 'metrics' in dir(self):
        self.metrics['onionprobe_state'].state('stopping')

    if 'controller' in dir(self):
        self.controller.close()

    # Terminate built-in Tor
    if 'tor' in dir(self):
        self.tor.kill()

time

OnionprobeTime

Onionprobe class with timing-related methods.

Source code in packages/onionprobe/time.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class OnionprobeTime:
    """
    Onionprobe class with timing-related methods.
    """

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

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

        return datetime.now()

    def wait(self, value):
        """
        Helper to wait some time

        :type  value: int
        :param value: Number of seconds to wait.
        """

        # Randomize if needed
        if self.get_config('randomize'):
            value = random.random() * value

        # Sleep, collecting metrics about it
        self.log('Waiting {} seconds...'.format(str(round(value))))
        self.metrics['onionprobe_wait_seconds'].set(value)
        self.metrics['onionprobe_state'].state('sleeping')
        time.sleep(value)

    def elapsed(self, init_time, verbose = False, label = ''):
        """
        Calculate the time elapsed since an initial time.

        :type  init_time: datetime.datetime
        :param init_time: Initial time.

        :type  verbose: bol
        :param verbose: If verbose is True, logs the elapsed time.
                        Defaults to False.

        :type  label: str
        :param label: A label to add in the elapsed time log message.
                      Only used if verbose is set to true.
                      Defaults to an empty string.

        :rtype: int
        :return: Number of elapsed time in seconds
        """

        # Calculate the elapsed time
        elapsed = (datetime.now() - init_time)

        # Log the elapsed time
        if verbose:
            if label != '':
                label = ' (' + str(label) + ')'

            self.log("Elapsed time" + label + ": " + str(elapsed))

        return timedelta.total_seconds(elapsed)

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

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

        return datetime.now().timestamp()

elapsed(init_time, verbose=False, label='')

Calculate the time elapsed since an initial time.

Parameters:

Name Type Description Default
init_time

Initial time.

 required
verbose

If verbose is True, logs the elapsed time. Defaults to False.

 False
label

A label to add in the elapsed time log message. Only used if verbose is set to true. Defaults to an empty string.

 ''

Returns:

Type Description
int

Number of elapsed time in seconds
Source code in packages/onionprobe/time.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def elapsed(self, init_time, verbose = False, label = ''):
    """
    Calculate the time elapsed since an initial time.

    :type  init_time: datetime.datetime
    :param init_time: Initial time.

    :type  verbose: bol
    :param verbose: If verbose is True, logs the elapsed time.
                    Defaults to False.

    :type  label: str
    :param label: A label to add in the elapsed time log message.
                  Only used if verbose is set to true.
                  Defaults to an empty string.

    :rtype: int
    :return: Number of elapsed time in seconds
    """

    # Calculate the elapsed time
    elapsed = (datetime.now() - init_time)

    # Log the elapsed time
    if verbose:
        if label != '':
            label = ' (' + str(label) + ')'

        self.log("Elapsed time" + label + ": " + str(elapsed))

    return timedelta.total_seconds(elapsed)

now()

Wrapper around datetime.now()

Returns:

Type Description
datetime.datetime

Current time.
Source code in packages/onionprobe/time.py 
32
33
34
35
36
37
38
39
40
def now(self):
    """
    Wrapper around datetime.now()

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

    return datetime.now()

timestamp()

Wrapper around datetime.now().timestamp()

Returns:

Type Description
datetime.datetime

Current time.
Source code in packages/onionprobe/time.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
def timestamp(self):
    """
    Wrapper around datetime.now().timestamp()

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

    return datetime.now().timestamp()

wait(value)

Helper to wait some time

Parameters:

Name Type Description Default
value

Number of seconds to wait.

 required
Source code in packages/onionprobe/time.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def wait(self, value):
    """
    Helper to wait some time

    :type  value: int
    :param value: Number of seconds to wait.
    """

    # Randomize if needed
    if self.get_config('randomize'):
        value = random.random() * value

    # Sleep, collecting metrics about it
    self.log('Waiting {} seconds...'.format(str(round(value))))
    self.metrics['onionprobe_wait_seconds'].set(value)
    self.metrics['onionprobe_state'].state('sleeping')
    time.sleep(value)

tls

OnionprobeTLS

Onionprobe class with TLS methods.

Source code in packages/onionprobe/tls.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
class OnionprobeTLS:
    """
    Onionprobe class with TLS methods.
    """

    def query_tls(self, endpoint, config, attempt = 1):
        """
        Tries a TLS connection to the endpoint and update metrics when needed.

        This method does not make any certificate verification upfront when
        connecting to the remote endpoint. This is on purpose, since this is
        just a test procedure to get TLS and certificate information.

        Certificate validity check is already done at OnionprobeHTTP.query_http().

        :type  endpoint: str
        :param endpoint: The endpoint name from the 'endpoints' instance config.

        :type  config: dict
        :param config: Endpoint configuration

        :type  attempt: int
        :param attempt: The current attempt used to determine the maximum
                        number of retries.

        :rtype: bool
        :return: True if the connection succeeded.
                 False on error.

        """

        tor_address = self.get_config('tor_address')
        socks_port  = self.get_config('socks_port')
        timeout     = self.get_config('tls_connect_timeout')
        port        = int(config['port']) if 'port' in config else 443
        exception   = None

        # Approach to use when always checking the certificate
        #context                = ssl.create_default_context()
        #context.check_hostname = True
        #context.verify_mode    = ssl.CERT_REQUIRED
        #valid_cert             = 1

        # Approach to use to retrieve whichever certificate, no matter whether it's valid or not
        context                = ssl.SSLContext()
        context.check_hostname = False
        context.verify_mode    = ssl.CERT_NONE
        valid_cert             = 1

        # Metric labels
        labels = {
                'name'    : endpoint,
                'address' : config['address'],
                'port'    : config['port'],
                }

        try:
            self.log('Trying to do a TLS connection to {} on port {} (attempt {})...'.format(
                config['address'], config['port'], attempt))

            with socks.create_connection(
                    (config['address'], port),
                    timeout=timeout, proxy_type=socks.SOCKS5,
                    proxy_addr=tor_address, proxy_port=socks_port, proxy_rdns=True) as sock:
                with context.wrap_socket(sock, server_hostname=config['address']) as tls:
                    result = True

                    self.log('TLS connection succeeded at {} on port {}'.format(
                            config['address'], config['port']))

                    if self.get_config('get_certificate_info'):
                        cert_result = self.get_certificate(endpoint, config, tls)

                    alpn        = tls.selected_alpn_protocol()
                    npn         = tls.selected_npn_protocol()
                    compression = tls.compression()
                    stats       = context.session_stats()
                    info        = {
                        'version'    : tls.version(),
                        'cipher'     : ' '.join([str(item) for item in tls.cipher()]),
                        'compression': '' if compression is None else str(compression),
                        'alpn'       : '' if alpn        is None else str(alpn),
                        'npn'        : '' if npn         is None else str(npn),
                        }

                    for item in stats:
                        info['session_' + item] = str(stats[item])

                    self.info_metric('onion_service_tls', info, labels)

                    # Requires Python 3.10+
                    if hasattr(context, 'security_level'):
                        self.set_metric('onion_service_tls_security_level', context.security_level, labels)

        except ssl.SSLZeroReturnError as e:
            result = False
            error  = e.reason

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'ssl_zero_return_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except ssl.SSLWantReadError as e:
            result = False
            error  = e.reason

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'ssl_want_read_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except ssl.SSLWantWriteError as e:
            result = False
            error  = e.reason

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'ssl_want_write_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except ssl.SSLSyscallError as e:
            result = False
            error  = e.reason

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'ssl_syscall_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except ssl.SSLEOFError as e:
            result = False
            error  = e.reason

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'ssl_eof_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        # This should never trigger since the TLS test does not check for
        # certificate validation.
        #except ssl.SSLCertVerificationError as e:
        #    result     = False
        #    error      = e.reason
        #    exception  = 'ssl_cert_verification_error'
        #    valid_cert = 0

        #    self.log(e, 'error')

        # Alias for ssl.CertificateVerificationError
        #except ssl.CertificateError as e:
        #    result    = False
        #    error     = e.reason
        #    exception = 'ssl_certificate_error'

        #    self.log(e, 'error')

        except ssl.SSLError as e:
            result = False
            error  = e.reason

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'ssl_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except socks.SOCKS5AuthError as e:
            result = False
            error  = e.socket.err

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'socks5_auth_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except socks.SOCKS5Error as e:
            result = False
            error  = e.socket.err

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'socks5_general_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except socks.HTTPError as e:
            result    = False
            error     = e.socket.err
            exception = 'http_error'

            self.log(e, 'error')

        except socks.GeneralProxyError as e:
            result = False
            error  = e.socket.err

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'general_proxy_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        except Exception as e:
            result = False

            # Do not use a fine grained exception metric here, but instead rely
            # on an existing metric used by other tests such as the HTTP
            #exception = 'generic_error'
            exception  = 'connection_error'

            self.log(e, 'error')

        finally:
            reachable = 0 if result is False else 1

            if result is False:
                retries = self.get_config('tls_connect_max_retries')

                # Try again until max retries is reached
                if attempt <= retries:
                    return self.query_tls(endpoint, config, attempt + 1)

            if exception is not None:
                # Count exceptions
                self.inc_metric('onion_service_' + exception + '_total', 1, labels)

                # Count errors
                self.inc_metric('onion_service_fetch_error_total', 1, labels)

                # Register the number attempts on metrics, but only in case of errors,
                # otherwse it may be redundant with what's already done at
                # This metrics may be too specific and can cause confusion with
                # OnionprobeHTTP.query_http()
                labels['reachable'] = reachable
                self.set_metric('onion_service_connection_attempts', attempt, labels)

            return result

query_tls(endpoint, config, attempt=1)

Tries a TLS connection to the endpoint and update metrics when needed.

This method does not make any certificate verification upfront when connecting to the remote endpoint. This is on purpose, since this is just a test procedure to get TLS and certificate information.

Certificate validity check is already done at OnionprobeHTTP.query_http().

Parameters:

Name Type Description Default
endpoint

The endpoint name from the 'endpoints' instance config.

 required
config

Endpoint configuration

 required
attempt

The current attempt used to determine the maximum number of retries.

 1

Returns:

Type Description
bool

True if the connection succeeded. False on error.
Source code in packages/onionprobe/tls.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def query_tls(self, endpoint, config, attempt = 1):
    """
    Tries a TLS connection to the endpoint and update metrics when needed.

    This method does not make any certificate verification upfront when
    connecting to the remote endpoint. This is on purpose, since this is
    just a test procedure to get TLS and certificate information.

    Certificate validity check is already done at OnionprobeHTTP.query_http().

    :type  endpoint: str
    :param endpoint: The endpoint name from the 'endpoints' instance config.

    :type  config: dict
    :param config: Endpoint configuration

    :type  attempt: int
    :param attempt: The current attempt used to determine the maximum
                    number of retries.

    :rtype: bool
    :return: True if the connection succeeded.
             False on error.

    """

    tor_address = self.get_config('tor_address')
    socks_port  = self.get_config('socks_port')
    timeout     = self.get_config('tls_connect_timeout')
    port        = int(config['port']) if 'port' in config else 443
    exception   = None

    # Approach to use when always checking the certificate
    #context                = ssl.create_default_context()
    #context.check_hostname = True
    #context.verify_mode    = ssl.CERT_REQUIRED
    #valid_cert             = 1

    # Approach to use to retrieve whichever certificate, no matter whether it's valid or not
    context                = ssl.SSLContext()
    context.check_hostname = False
    context.verify_mode    = ssl.CERT_NONE
    valid_cert             = 1

    # Metric labels
    labels = {
            'name'    : endpoint,
            'address' : config['address'],
            'port'    : config['port'],
            }

    try:
        self.log('Trying to do a TLS connection to {} on port {} (attempt {})...'.format(
            config['address'], config['port'], attempt))

        with socks.create_connection(
                (config['address'], port),
                timeout=timeout, proxy_type=socks.SOCKS5,
                proxy_addr=tor_address, proxy_port=socks_port, proxy_rdns=True) as sock:
            with context.wrap_socket(sock, server_hostname=config['address']) as tls:
                result = True

                self.log('TLS connection succeeded at {} on port {}'.format(
                        config['address'], config['port']))

                if self.get_config('get_certificate_info'):
                    cert_result = self.get_certificate(endpoint, config, tls)

                alpn        = tls.selected_alpn_protocol()
                npn         = tls.selected_npn_protocol()
                compression = tls.compression()
                stats       = context.session_stats()
                info        = {
                    'version'    : tls.version(),
                    'cipher'     : ' '.join([str(item) for item in tls.cipher()]),
                    'compression': '' if compression is None else str(compression),
                    'alpn'       : '' if alpn        is None else str(alpn),
                    'npn'        : '' if npn         is None else str(npn),
                    }

                for item in stats:
                    info['session_' + item] = str(stats[item])

                self.info_metric('onion_service_tls', info, labels)

                # Requires Python 3.10+
                if hasattr(context, 'security_level'):
                    self.set_metric('onion_service_tls_security_level', context.security_level, labels)

    except ssl.SSLZeroReturnError as e:
        result = False
        error  = e.reason

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'ssl_zero_return_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except ssl.SSLWantReadError as e:
        result = False
        error  = e.reason

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'ssl_want_read_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except ssl.SSLWantWriteError as e:
        result = False
        error  = e.reason

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'ssl_want_write_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except ssl.SSLSyscallError as e:
        result = False
        error  = e.reason

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'ssl_syscall_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except ssl.SSLEOFError as e:
        result = False
        error  = e.reason

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'ssl_eof_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    # This should never trigger since the TLS test does not check for
    # certificate validation.
    #except ssl.SSLCertVerificationError as e:
    #    result     = False
    #    error      = e.reason
    #    exception  = 'ssl_cert_verification_error'
    #    valid_cert = 0

    #    self.log(e, 'error')

    # Alias for ssl.CertificateVerificationError
    #except ssl.CertificateError as e:
    #    result    = False
    #    error     = e.reason
    #    exception = 'ssl_certificate_error'

    #    self.log(e, 'error')

    except ssl.SSLError as e:
        result = False
        error  = e.reason

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'ssl_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except socks.SOCKS5AuthError as e:
        result = False
        error  = e.socket.err

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'socks5_auth_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except socks.SOCKS5Error as e:
        result = False
        error  = e.socket.err

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'socks5_general_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except socks.HTTPError as e:
        result    = False
        error     = e.socket.err
        exception = 'http_error'

        self.log(e, 'error')

    except socks.GeneralProxyError as e:
        result = False
        error  = e.socket.err

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'general_proxy_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    except Exception as e:
        result = False

        # Do not use a fine grained exception metric here, but instead rely
        # on an existing metric used by other tests such as the HTTP
        #exception = 'generic_error'
        exception  = 'connection_error'

        self.log(e, 'error')

    finally:
        reachable = 0 if result is False else 1

        if result is False:
            retries = self.get_config('tls_connect_max_retries')

            # Try again until max retries is reached
            if attempt <= retries:
                return self.query_tls(endpoint, config, attempt + 1)

        if exception is not None:
            # Count exceptions
            self.inc_metric('onion_service_' + exception + '_total', 1, labels)

            # Count errors
            self.inc_metric('onion_service_fetch_error_total', 1, labels)

            # Register the number attempts on metrics, but only in case of errors,
            # otherwse it may be redundant with what's already done at
            # This metrics may be too specific and can cause confusion with
            # OnionprobeHTTP.query_http()
            labels['reachable'] = reachable
            self.set_metric('onion_service_connection_attempts', attempt, labels)

        return result

tor

OnionprobeTor

Onionprobe class with Tor-related methods.

Source code in packages/onionprobe/tor.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
class OnionprobeTor:
    """
    Onionprobe class with Tor-related methods.
    """

    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)

        # Workaround for https://github.com/torproject/stem/issues/112
        if self.get_config('log_level') != 'debug':
            stem.util.log.get_logger().propagate = False

        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

    def initialize_tor_auth(self):
        """
        Initialize an authenticated Tor control connection

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

        if 'controller' not in dir(self):
            self.log("Unable to find a Tor control connection", "critical")

            return False

        # Evaluate control password only after we're sure that a Tor
        # process is running in the case that 'launch_tor' is True
        control_password = self.get_config('control_password')

        if control_password is False:
            # First try to authenticate without a password
            try:
                self.controller.authenticate()

            # Then fallback to ask for a password
            except stem.connection.MissingPassword:
                import getpass
                control_password = getpass.getpass("Controller password: ")

                try:
                    self.controller.authenticate(password=control_password)
                except stem.connection.PasswordAuthFailed:
                    self.log("Unable to authenticate, password is incorrect", "critical")

                    return False

        else:
            try:
                self.controller.authenticate(password=control_password)
            except stem.connection.PasswordAuthFailed:
                self.log("Unable to authenticate, password is incorrect", "critical")

                return False

        return True

    def initialize_listeners(self):
        """
        Initialize Tor event listeners
        """

        # Stream management
        # See https://stem.torproject.org/tutorials/to_russia_with_love.html
        if self.get_config('new_circuit'):
            self.controller.set_conf('__LeaveStreamsUnattached', '1')
            self.controller.add_event_listener(self.new_circuit, stem.control.EventType.STREAM)

            self.circuit_id = None

        # Add listener for Onion Services descriptors
        self.controller.add_event_listener(self.hsdesc_event, stem.control.EventType.HS_DESC)

    #
    # Tor related logic
    #

    def gen_control_password(self):
        """
        Generates a random password

        :rtype: str
        :return: A random password between 22 and 32 bytes
        """

        import secrets

        return secrets.token_urlsafe(random.randrange(22, 32))

    def hash_password(self, password):
        """
        Produce a hashed password in the format used by HashedControlPassword

        It currently relies on spawning a "tor --hash-password" process so it suffering
        from the security issue of temporarily exposing the unhashed password in the
        operating system's list of running processes.

        :type  password: str
        :param password: A password to be hashed

        :rtype: str
        :return: The hashed password
        """

        import subprocess

        tor    = shutil.which('tor')
        result = subprocess.check_output([tor, '--quiet', '--hash-password', password], text=True)

        return result

    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

    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)

gen_control_password()

Generates a random password

Returns:

Type Description
str

A random password between 22 and 32 bytes
Source code in packages/onionprobe/tor.py 
143
144
145
146
147
148
149
150
151
152
153
def gen_control_password(self):
    """
    Generates a random password

    :rtype: str
    :return: A random password between 22 and 32 bytes
    """

    import secrets

    return secrets.token_urlsafe(random.randrange(22, 32))

hash_password(password)

Produce a hashed password in the format used by HashedControlPassword

It currently relies on spawning a "tor --hash-password" process so it suffering from the security issue of temporarily exposing the unhashed password in the operating system's list of running processes.

Parameters:

Name Type Description Default
password

A password to be hashed

 required

Returns:

Type Description
str

The hashed password
Source code in packages/onionprobe/tor.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def hash_password(self, password):
    """
    Produce a hashed password in the format used by HashedControlPassword

    It currently relies on spawning a "tor --hash-password" process so it suffering
    from the security issue of temporarily exposing the unhashed password in the
    operating system's list of running processes.

    :type  password: str
    :param password: A password to be hashed

    :rtype: str
    :return: The hashed password
    """

    import subprocess

    tor    = shutil.which('tor')
    result = subprocess.check_output([tor, '--quiet', '--hash-password', password], text=True)

    return result

initialize_listeners()

Initialize Tor event listeners

Source code in packages/onionprobe/tor.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def initialize_listeners(self):
    """
    Initialize Tor event listeners
    """

    # Stream management
    # See https://stem.torproject.org/tutorials/to_russia_with_love.html
    if self.get_config('new_circuit'):
        self.controller.set_conf('__LeaveStreamsUnattached', '1')
        self.controller.add_event_listener(self.new_circuit, stem.control.EventType.STREAM)

        self.circuit_id = None

    # Add listener for Onion Services descriptors
    self.controller.add_event_listener(self.hsdesc_event, stem.control.EventType.HS_DESC)

initialize_tor()

Initialize Tor control connection

Returns:

Type Description
bol

True if initialization is successful, False on error
Source code in packages/onionprobe/tor.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
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)

    # Workaround for https://github.com/torproject/stem/issues/112
    if self.get_config('log_level') != 'debug':
        stem.util.log.get_logger().propagate = False

    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

initialize_tor_auth()

Initialize an authenticated Tor control connection

Returns:

Type Description
bol

True if initialization is successful, False on error
Source code in packages/onionprobe/tor.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def initialize_tor_auth(self):
    """
    Initialize an authenticated Tor control connection

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

    if 'controller' not in dir(self):
        self.log("Unable to find a Tor control connection", "critical")

        return False

    # Evaluate control password only after we're sure that a Tor
    # process is running in the case that 'launch_tor' is True
    control_password = self.get_config('control_password')

    if control_password is False:
        # First try to authenticate without a password
        try:
            self.controller.authenticate()

        # Then fallback to ask for a password
        except stem.connection.MissingPassword:
            import getpass
            control_password = getpass.getpass("Controller password: ")

            try:
                self.controller.authenticate(password=control_password)
            except stem.connection.PasswordAuthFailed:
                self.log("Unable to authenticate, password is incorrect", "critical")

                return False

    else:
        try:
            self.controller.authenticate(password=control_password)
        except stem.connection.PasswordAuthFailed:
            self.log("Unable to authenticate, password is incorrect", "critical")

            return False

    return True

launch_tor()

Launch a built-in Tor process

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

Source code in packages/onionprobe/tor.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
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

new_circuit(stream)

Setup a fresh Tor circuit for new streams

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

Parameters:

Name Type Description Default
stream

Stream event

 required
Source code in packages/onionprobe/tor.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
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)