Source code for darc.proxy.i2p

# -*- coding: utf-8 -*-
# pylint: disable=ungrouped-imports
"""I2P Proxy

The :mod:`darc.proxy.i2p` module contains the auxiliary functions
around managing and processing the I2P proxy.


import base64
import getpass
import os
import platform
import re
import shlex
import signal
import subprocess  # nosec: B404
from typing import TYPE_CHECKING, cast

import requests
import selenium.webdriver.common.proxy as selenium_proxy

from darc.const import CHECK, DARC_USER, DEBUG, PATH_DB
from darc.error import I2PBootstrapFailed, UnsupportedPlatform
from import parse_link, urljoin
from darc.logging import DEBUG as LOG_DEBUG
from darc.logging import ERROR as LOG_ERROR
from darc.logging import INFO as LOG_INFO
from darc.logging import VERBOSE as LOG_VERBOSE
from darc.logging import WARNING as LOG_WARNING
from darc.logging import logger
from darc.parse import _check, get_content_type

    from io import IO  # type: ignore[attr-defined] # pylint: disable=no-name-in-module
    from signal import Signals  # pylint: disable=no-name-in-module
    from subprocess import Popen  # nosec: B404
    from types import FrameType
    from typing import List, NoReturn, Optional, Union

    import as darc_link  # Link
    from darc._typing import File

# I2P args
I2P_ARGS = shlex.split(os.getenv('I2P_ARGS', ''))

# bootstrap wait
BS_WAIT = float(os.getenv('I2P_WAIT', '90'))

# I2P port
I2P_PORT = os.getenv('I2P_PORT', '4444')

# I2P bootstrap retry
I2P_RETRY = int(os.getenv('I2P_RETRY', '3'))

# proxy
    # c.f.
    'http':  f'http://localhost:{I2P_PORT}',
    'https': f'http://localhost:{I2P_PORT}',
I2P_SELENIUM_PROXY = selenium_proxy.Proxy()
I2P_SELENIUM_PROXY.proxyType = selenium_proxy.ProxyType.MANUAL
I2P_SELENIUM_PROXY.http_proxy = f'http://localhost:{I2P_PORT}'
I2P_SELENIUM_PROXY.ssl_proxy = f'http://localhost:{I2P_PORT}'

# manage I2P through darc?
_MNG_I2P = bool(int(os.getenv('DARC_I2P', '1')))

# I2P bootstrapped flag
_I2P_BS_FLAG = not _MNG_I2P
# I2P daemon process
_I2P_PROC = None
# I2P bootstrap args
_unsupported = False
if getpass.getuser() == 'root':
    _system = platform.system()
    if _system in ['Linux', 'Darwin']:
        _I2P_ARGS = ['su', '-', DARC_USER, 'i2prouter', 'start']
        _unsupported = True
        _I2P_ARGS = []
    _I2P_ARGS = ['i2prouter', 'start']

if _unsupported:
    if DEBUG:
        logger.debug('-*- FREENET PROXY -*-')
        logger.pline(LOG_ERROR, 'unsupported system: %s', platform.system())
        logger.pline(LOG_DEBUG, logger.horizon)
    logger.plog(LOG_DEBUG, '-*- FREENET PROXY -*-', object=_I2P_ARGS)

# I2P link regular expression
I2P_REGEX = re.compile(r'.*\.i2p', re.IGNORECASE)

[docs]def launch_i2p() -> 'Popen[bytes]': """Launch I2P process. See Also: This function mocks the behaviour of :func:`stem.process.launch_tor`. """ i2p_process = None try: i2p_process = subprocess.Popen( # pylint: disable=consider-using-with # nosec _I2P_ARGS, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) def timeout_handlet(signum: 'Optional[Union[int, Signals]]' = None, frame: 'Optional[FrameType]' = None) -> 'NoReturn': raise OSError('reached a %i second timeout without success' % BS_WAIT) signal.signal(signal.SIGALRM, timeout_handlet) signal.setitimer(signal.ITIMER_REAL, BS_WAIT) while True: init_line = cast( 'IO[bytes]', i2p_process.stdout ).readline().decode('utf-8', 'replace').strip() logger.pline(LOG_VERBOSE, init_line) if not init_line: raise OSError('Process terminated: Timed out') if 'running: PID:' in init_line: return i2p_process if 'I2P Service is already running.' in init_line: return i2p_process except BaseException: if i2p_process is not None: i2p_process.kill() # don't leave a lingering process i2p_process.wait() raise finally: signal.alarm(0) # stop alarm
[docs]def _i2p_bootstrap() -> None: """I2P bootstrap. The bootstrap arguments are defined as :data:`~darc.proxy.i2p._I2P_ARGS`. Raises: subprocess.CalledProcessError: If the return code of :data:`~darc.proxy.i2p._I2P_PROC` is non-zero. See Also: * :func:`darc.proxy.i2p.i2p_bootstrap` * :func:`darc.proxy.i2p.launch_i2p` * :data:`darc.proxy.i2p.BS_WAIT` * :data:`darc.proxy.i2p._I2P_BS_FLAG` * :data:`darc.proxy.i2p._I2P_PROC` """ global _I2P_BS_FLAG, _I2P_PROC # pylint: disable=global-statement # launch I2P process _I2P_PROC = launch_i2p() # update flag _I2P_BS_FLAG = True
[docs]def i2p_bootstrap() -> None: """Bootstrap wrapper for I2P. The function will bootstrap the I2P proxy. It will retry for :data:`~darc.proxy.i2p.I2P_RETRY` times in case of failure. Also, it will **NOT** re-bootstrap the proxy as is guaranteed by :data:`~darc.proxy.i2p._I2P_BS_FLAG`. Warns: I2PBootstrapFailed: If failed to bootstrap I2P proxy. Raises: :exc:`UnsupportedPlatform`: If the system is not supported, i.e. not macOS or Linux. See Also: * :func:`darc.proxy.i2p._i2p_bootstrap` * :data:`darc.proxy.i2p.I2P_RETRY` * :data:`darc.proxy.i2p._I2P_BS_FLAG` """ if _unsupported: raise UnsupportedPlatform(f'unsupported system: {platform.system()}') # don't re-bootstrap if _I2P_BS_FLAG: return'-*- I2P Bootstrap -*-') for _ in range(I2P_RETRY+1): try: _i2p_bootstrap() break except Exception: if DEBUG: logger.ptb('[Error bootstraping I2P proxy]') logger.pexc(LOG_WARNING, category=I2PBootstrapFailed, line='i2p_bootstrap()') logger.pline(LOG_INFO, logger.horizon)
[docs]def get_hosts(link: 'darc_link.Link') -> 'Optional[File]': """Read ``hosts.txt``. Args: link: Link object to read ``hosts.txt``. Returns: * If ``hosts.txt`` exists, return the data from ``hosts.txt``. * ``path`` -- relative path from ``hosts.txt`` to root of data storage :data:`~darc.const.PATH_DB`, ``<proxy>/<scheme>/<hostname>/hosts.txt`` * ``data`` -- *base64* encoded content of ``hosts.txt`` * If not, return :data:`None`. See Also: * :func:`darc.submit.submit_new_host` * :func:`darc.proxy.i2p.save_hosts` """ path = os.path.join(link.base, 'hosts.txt') if not os.path.isfile(path): return None with open(path, 'rb') as file: content = return { 'path': os.path.relpath(path, PATH_DB), 'data': base64.b64encode(content).decode(), }
[docs]def have_hosts(link: 'darc_link.Link') -> 'Optional[str]': """Check if ``hosts.txt`` already exists. Args: link: Link object to check if ``hosts.txt`` already exists. Returns: * If ``hosts.txt`` exists, return the path to ``hosts.txt``, i.e. ``<root>/<proxy>/<scheme>/<hostname>/hosts.txt``. * If not, return :data:`None`. """ # <proxy>/<scheme>/<host>/hosts.txt path = os.path.join(link.base, 'hosts.txt') return path if os.path.isfile(path) else None
[docs]def save_hosts(link: 'darc_link.Link', text: str) -> str: """Save ``hosts.txt``. Args: link: Link object of ``hosts.txt``. text: Content of ``hosts.txt``. Returns: Saved path to ``hosts.txt``, i.e. ``<root>/<proxy>/<scheme>/<hostname>/hosts.txt``. See Also: * :func:`` """ path = os.path.join(link.base, 'hosts.txt') root = os.path.split(path)[0] os.makedirs(root, exist_ok=True) with open(path, 'w') as file: print(f'# {link.url}', file=file) file.write(text) return path
[docs]def read_hosts(link: 'darc_link.Link', text: str, check: bool = CHECK) -> 'List[darc_link.Link]': """Read ``hosts.txt``. Args: link: Link object to fetch for its ``hosts.txt``. text: Content of ``hosts.txt``. check: If perform checks on extracted links, default to :data:`~darc.const.CHECK`. Returns: List of links extracted. """ temp_list = [] for line in filter(None, map(lambda s: s.strip(), text.splitlines())): if line.startswith('#'): continue host = line.split('=', maxsplit=1)[0] if I2P_REGEX.fullmatch(host) is None: continue temp_list.append(parse_link(f'http://{host}', backref=link)) if check: return _check(temp_list) return temp_list
[docs]def fetch_hosts(link: 'darc_link.Link', force: bool = False) -> None: """Fetch ``hosts.txt``. Args: link: Link object to fetch for its ``hosts.txt``. force: Force refetch ``hosts.txt``. Returns: Content of the ``hosts.txt`` file. """ if force: logger.warning('[HOSTS] Force refetch %s', link.url) hosts_path = None if force else have_hosts(link) if hosts_path is not None: logger.warning('[HOSTS] Cached %s', link.url) # pylint: disable=no-member with open(hosts_path) as hosts_file: hosts_text = else: from darc.requests import i2p_session # pylint: disable=import-outside-toplevel hosts_link = parse_link(urljoin(link.url, '/hosts.txt'), backref=link)'[HOSTS] Subscribing %s', hosts_link.url) with i2p_session() as session: try: response = session.get(hosts_link.url) except requests.RequestException: logger.pexc(message=f'[HOSTS] Failed on {hosts_link.url}') return if not response.ok: logger.error('[HOSTS] Failed on %s [%d]', hosts_link.url, response.status_code) return ct_type = get_content_type(response) if ct_type not in ['text/text', 'text/plain']: logger.error('[HOSTS] Unresolved content type on %s (%s)', hosts_link.url, ct_type) return hosts_text = response.text save_hosts(hosts_link, hosts_text)'[HOSTS] Subscribed %s', hosts_link.url) from darc.db import save_requests # pylint: disable=import-outside-toplevel # add link to queue save_requests(read_hosts(link, hosts_text))