emdee
/
stem_examples
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

tests

This commit is contained in:
emdee@spm.plastiras.org 2024-01-16 18:10:43 +00:00
parent 42cf62f623
commit 72010d8f8d
9 changed files with 192 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -4,6 +4,7 @@ __pycache__/
*.py[cod]
*$py.class
#*
*~
*.dst

21
Makefile
View File

@ -1,8 +1,9 @@
LOCAL_DOCTEST=/usr/local/bin/toxcore_run_doctest3.bash
gggggPREFIX=/usr/local
PYTHON_EXE_MSYS=${PREFIX}/bin/python3.sh
LOCAL_DOCTEST=${PREFIX}/bin/toxcore_run_doctest3.bash
DOCTEST=${LOCAL_DOCTEST}
MOD=stem_examples
check::
sh python3.sh -c "import ${MOD}"
@ -15,9 +16,23 @@ rsync::
pyi::
echo FixMe
test::
export TOR_CONTROLLER_PASSWORD=${PASS}
${PYTHON_EXE_MSYS} src/${MOD}/check_digests.py
# broken because this site fails: http://128.31.0.39:9131/tor/status-vote
# ${PYTHON_EXE_MSYS} src/${MOD}/compare_flags.py
# cant use from make: waits for the cmdline to to terminate
# TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/exit_used.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/introduction_points.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/list_circuits.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/mappaddress.py
#hangs TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/outdated_relays.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/relay_connections.py
#wrosk TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/tor_bootstrap_check.py
doctest:
export PYTHONPATH=${PWD}
${DOCTEST} ${MOD].txt
${DOCTEST} ${MOD}.txt
veryclean:: clean
rm -rf build dist __pycache__ .pylint.err .pylint.out

69
src/stem_examples/check_digests.py
View File

@ -15,7 +15,8 @@ import sys
import contextlib
import logging
from tor_controller import set_socks_proxy
from stem_examples.tor_controller import set_socks_proxy
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
@ -56,6 +57,7 @@ def download_descriptors(fingerprint):
)
def iMain(lArgs=None):
global LOG
# set_socks_proxy()
iRetval = 0
if lArgs is None:
@ -64,7 +66,7 @@ def iMain(lArgs=None):
lArgs = [fingerprint]
for fingerprint in lArgs:
log.INFO(f"checking digests of fp={fp}")
LOG.info(f"checking digests of fp={fp}")
if not stem.util.tor_tools.is_valid_fingerprint(fingerprint):
LOG.error("'%s' is not a valid relay fingerprint" % fingerprint)
iRetval += 1
@ -91,37 +93,40 @@ def iMain(lArgs=None):
return iRetval
if __name__ == '__main__':
LOG.setLevel(logging.DEBUG)
try:
import stem.descriptor.remote
import stem.util.tor_tools
logging.getLogger('stem').setLevel(20)
# bizarre uncatchable stem error
import stem.response.protocolinfo
import stem.response.mapaddress
import stem.response.getconf
import stem.response.getinfo
import stem.response.authchallenge
# 'tuple' object has no attribute 'endswith'
if len(sys.argv) > 1:
lArgs = sys.argv[1:]
LOG.info(f"Getting some {len(lArgs)}")
if os.environ.get('DEBUG', ''):
log_level = 10
else:
sKNOWN_ONION = 'facebookwkhpilnemxj7asaniu7vnjjbiltxjqhye3mhbshg7kx5tfyd', # facebook
LOG.info("Getting some FPs from a sKNOWN_ONION")
from stem_examples.introduction_points import lMain
# with ignoreStdout():
lArgs = lMain([sKNOWN_ONION])
LOG.debug(f"Got {len(lArgs)} FPs from a sKNOWN_ONION")
log_level = 20
vsetup_logging(LOG, log_level)
try:
logging.getLogger('stem').setLevel(20)
import stem.descriptor.remote
import stem.util.tor_tools
# bizarre uncatchable stem error
import stem.response.protocolinfo
import stem.response.mapaddress
import stem.response.getconf
import stem.response.getinfo
import stem.response.authchallenge
# 'tuple' object has no attribute 'endswith'
i = iMain(lArgs)
except KeyboardInterrupt as e:
i = 0
except Exception as e:
LOG.exception(f"Exception in iMain {e}")
i = 1
sys.exit(i)
if len(sys.argv) > 1:
lArgs = sys.argv[1:]
LOG.info(f"Getting some {len(lArgs)}")
else:
sKNOWN_ONION = 'facebookwkhpilnemxj7asaniu7vnjjbiltxjqhye3mhbshg7kx5tfyd' # facebook
LOG.info("Getting some FPs from a sKNOWN_ONION")
from stem_examples.introduction_points import lMain
with ignoreStdout():
lArgs = lMain([sKNOWN_ONION])
LOG.info(f"Got {len(lArgs)} FPs from a sKNOWN_ONION")
i = iMain(lArgs)
except KeyboardInterrupt as e:
i = 0
except Exception as e:
LOG.exception(f"Exception in iMain {e}")
i = 1
sys.exit(i)

3
src/stem_examples/compare_flags.py
View File

@ -4,6 +4,9 @@ __doc__ = """
Compares the votes of two directory authorities, in this case moria1
and maatuska, with a special interest in the 'Running' flag.
broken because this site fails:
http://128.31.0.39:9131/tor/status-vote
https://stem.torproject.org/tutorials/examples/compare_flags.html
"""

42
src/stem_examples/introduction_points.py
View File

@ -26,6 +26,7 @@ import logging
from stem.control import Controller
from stem_examples.tor_controller import get_controller
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
@ -40,47 +41,52 @@ def iMain(lArgs=None):
return -1
return 0
def lMain(lArgs=None):
def lMain(lArgs=None, timeout=None):
lRetval = []
if lArgs is None:
if not lArgs:
lArgs = lKNOWN_ONIONS
try:
if os.path.exists('/run/tor/control'):
controller = get_controller(unix='/run/tor/control')
else:
controller = get_controller(port=9051)
password = os.environ.get('TOR_CONTROLLER_PASSWORD')
controller.authenticate(password)
if False and os.path.exists('/run/tor/control'):
controller = get_controller(password=password, unix='/run/tor/control')
else:
controller = get_controller(password=password, port=9051)
for elt in lArgs:
desc = controller.get_hidden_service_descriptor(elt, await_result=True, timeout=None)
print(f"{desc} get_hidden_service_descriptor\n")
desc = controller.get_hidden_service_descriptor(elt,
await_result=True,
timeout=timeout)
LOG.info(f"{desc} get_hidden_service_descriptor\n")
l = desc.introduction_points()
if l:
print(f"{elt} NO introduction points\n")
if not l:
LOG.warn(f"{elt} NO introduction points {l}\n")
continue
print(f"{elt} introduction points are...\n")
LOG.info(f"{elt} introduction points are...\n")
for introduction_point in l:
lRetval += [introduction_point]
print(' %s:%s => %s' % (introduction_point.address,
LOG.info(' %s:%s => %s' % (introduction_point.address,
introduction_point.port,
introduction_point.identifier))
except Exception as e:
print(e)
LOG.exception(f"Exception: {e}")
finally:
del controller
return lRetval
if __name__ == '__main__':
LOG.setLevel(logging.INFO)
if os.environ.get('DEBUG', ''):
log_level = 10
else:
log_level = 20
vsetup_logging(LOG, log_level)
try:
i = iMain(sys.argv[1:])
l = lMain(sys.argv[1:])
if l: print(l)
i = 0
except KeyboardInterrupt as e:
i = 0
except Exception as e:
i = 1
sys.exit(i)

23
src/stem_examples/mappaddress.py
View File

@ -8,22 +8,23 @@ import os
from stem import StreamStatus
from stem.control import EventType, Controller
from tor_controller import set_socks_proxy
from stem_examples.tor_controller import get_controller
from stem_examples.stem_utils import vsetup_logging
global LOG
import logging
LOG = logging.getLogger('map_address')
LOG = logging.getLogger()
def sMapaddressResolv(target, iPort=9051):
try:
if os.path.exists('/run/tor/control'):
controller = get_controller(unix='/run/tor/control')
else:
controller = get_controller(port=9051)
password = os.environ.get('TOR_CONTROLLER_PASSWORD')
controller.authenticate(password)
if os.path.exists('/run/tor/control'):
controller = get_controller(password=password, unix='/run/tor/control')
else:
controller = get_controller(password=password, port=9051)
map_dict = {"0.0.0.0": target}
map_ret = controller.map_address(map_dict)
@ -33,9 +34,15 @@ def sMapaddressResolv(target, iPort=9051):
LOG.exception(e)
if __name__ == '__main__':
if os.environ.get('DEBUG', ''):
log_level = 10
else:
log_level = 20
LOG = logging.getLogger()
vsetup_logging(LOG, log_level)
if len(sys.argv) < 2:
target = "l2ct3xnuaiwwtoybtn46qp2av4ndxcguwupzyv6xrsmnwi647vvmwtqd"
else:
target = sys.argv[1]
print(sMapaddressResolv(target))
LOG.info(sMapaddressResolv(target))

17
src/stem_examples/outdated_relays.py
View File

@ -24,7 +24,7 @@ def iMain():
downloader = DescriptorDownloader(use_mirrors=True)
count, with_contact = 0, 0
elts = downloader.get_server_descriptors()
print(f"Checking for outdated relays len server_descriptors={len(list(elts))}...")
LOG.info(f"Checking for outdated relays len server_descriptors={len(list(elts))}...")
print("")
for desc in elts:
@ -32,13 +32,22 @@ def iMain():
count += 1
if desc.contact:
print(' %-15s %s' % (desc.tor_version, desc.contact.decode("utf-8", "replace")))
LOG.info(' %-15s %s' % (desc.tor_version, desc.contact.decode("utf-8", "replace")))
with_contact += 1
print("")
print("%i outdated relays found, %i had contact information" % (count, with_contact))
LOG.info("%i outdated relays found, %i had contact information" % (count, with_contact))
# http://vt5hknv6sblkgf22.onion/tutorials/examples/outdated_relays.htmlhttp://vt5hknv6sblkgf22.onion/tutorials/examples/outdated_relays.html
return 0
if __name__ == '__main__':
sys.exit( iMain())
LOG.setLevel(logging.INFO)
try:
l = iMain(sys.argv[1:])
if l: print(l)
i = 0
except KeyboardInterrupt as e:
i = 0
except Exception as e:
i = 1
sys.exit(i)

34
src/stem_examples/relay_connections.py
View File

@ -28,7 +28,6 @@ from stem_examples.tor_controller import get_controller
global LOG
import logging
LOG = logging.getLogger('relay_cons')
HEADER_LINE = " {version} uptime: {uptime} flags: {flags}\n"
@ -52,15 +51,12 @@ def iMain(lArgs=None):
parser.add_argument("--resolver", help="default: autodetected")
args = parser.parse_args(lArgs)
control_port = int(args.ctrlport) if args.ctrlport else 'default'
if os.path.exists('/run/tor/control'):
controller = get_controller(unix='/run/tor/control')
else:
controller = get_controller(port=control_port)
# controller = stem.connection.connect(control_port = ('127.0.0.1', control_port))
password = os.environ.get('TOR_CONTROLLER_PASSWORD')
controller.authenticate(password)
control_port = int(args.ctrlport) if args.ctrlport else 'default'
if False and os.path.exists('/run/tor/control'):
controller = get_controller(password=password, unix='/run/tor/control')
else:
controller = get_controller(password=password, port=control_port)
if not controller:
return 1
@ -70,7 +66,7 @@ def iMain(lArgs=None):
version = str(controller.get_version()).split()[0],
uptime = stem.util.str_tools.short_time_label(time.time() - stem.util.system.start_time(pid))
print(HEADER_LINE.format(
LOG.info(HEADER_LINE.format(
version=version,
uptime=uptime,
flags = ', '.join(desc.flags if desc else ['none']),
@ -159,5 +155,21 @@ def iMain(lArgs=None):
if __name__ == '__main__':
iMain()
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
if os.environ.get('DEBUG', ''):
log_level = 10
else:
log_level = 20
vsetup_logging(LOG, log_level)
LOG.setLevel(logging.DEBUG)
try:
l = iMain()
if l: print(l)
i = 0
except KeyboardInterrupt as e:
i = 0
except Exception as e:
i = 1
sys.exit(i)

58
src/stem_examples/stem_utils.py Normal file
View File

@ -0,0 +1,58 @@
# -*-mode: python; py-indent-offset: 2; indent-tabs-mode: nil; coding: utf-8-unix -*-
import sys
import logging
try:
# if 'COLOREDLOGS_LEVEL_STYLES' not in os.environ:
# os.environ['COLOREDLOGS_LEVEL_STYLES'] = 'spam=22;debug=28;verbose=34;notice=220;warning=202;success=118,bold;error=124;critical=background=red'
# https://pypi.org/project/coloredlogs/
import coloredlogs
except ImportError:
coloredlogs = False
def vsetup_logging(theLOG, log_level, logfile='', stream=sys.stdout) -> None:
global LOG
LOG = theLOG
add = True
logging._defaultFormatter = logging.Formatter(datefmt='%m-%d %H:%M:%S')
logging._defaultFormatter.default_time_format = '%m-%d %H:%M:%S'
logging._defaultFormatter.default_msec_format = ''
if logfile:
add = logfile.startswith('+')
sub = logfile.startswith('-')
if add or sub:
logfile = logfile[1:]
kwargs['filename'] = logfile
if coloredlogs:
coloredlogs.DEFAULT_LEVEL_STYLES['info']=dict(color='white',bold=True)
coloredlogs.DEFAULT_LEVEL_STYLES['debug']=dict(color='cyan')
coloredlogs.DEFAULT_LEVEL_STYLES['warn']=dict(color='yellow',bold=True)
coloredlogs.DEFAULT_LEVEL_STYLES['error']=dict(color='red',bold=True)
coloredlogs.DEFAULT_FIELD_STYLES['levelname=']=dict(color='green', bold=True),
# https://pypi.org/project/coloredlogs/
aKw = dict(level=log_level,
logger=LOG,
stream=stream,
fmt='%(levelname)s %(message)s',
isatty=True,
milliseconds=False,
)
coloredlogs.install(**aKw)
if logfile:
oHandler = logging.FileHandler(logfile)
LOG.addHandler(oHandler)
LOG.debug(f"Setting coloured log_level to {log_level}")
else:
kwargs = dict(level=log_level,
force=True,
format='%(levelname)s %(message)s')
logging.basicConfig(**kwargs)
if add and logfile:
oHandler = logging.StreamHandler(stream)
LOG.addHandler(oHandler)
LOG.debug(f"SSetting log_level to {log_level}")