A dropin replacement for apache-users.pl written in Python 3

apache-users

#!/usr/bin/env python3
import sys
import threading
import argparse
import requests
import concurrent.futures
import logging
from urllib.parse import unquote

logging.basicConfig(format="%(asctime)s[%(levelname)8s][%(module)s] %(message)s", datefmt="[%m/%d/%Y %H:%M:%S]")
logger = logging.getLogger(__name__)

parser = argparse.ArgumentParser(description="Enumerate usernames on systems with Apache UserDir module", add_help=False)

parser.add_argument("--help", action="help", help="Show this help message and exit")
parser.add_argument("-h", dest="host", help="Address of remote host", required=True, metavar="host")
parser.add_argument("-l", dest="list", help="List of names to scan", required=True, type=argparse.FileType("r"),  metavar="list")
parser.add_argument("-p", dest="port", type=int, help="Port number", default=80, metavar="port")
parser.add_argument("-s", dest="ssl", action="store_true", help="Enable SSL support", default=False)
parser.add_argument("-e", dest="error_code", type=int, help="HTTP Error Code", default=403, metavar="http code")
parser.add_argument("-t", dest="threads", type=int, help="Number of threads", default=10, metavar="threads")
parser.add_argument("-w", dest="timeout", type=int, help="Request timeout", default=30, metavar="timeout")
parser.add_argument("-v", dest="verbosity", action="count", help="Increase verbosity.")
args = parser.parse_args()

# Set root logging level
if args.verbosity is not None:
    logger = logging.getLogger()
    if args.verbosity == 1:
        logger.setLevel(logging.INFO)
    else:
        logger.setLevel(logging.DEBUG)

words = args.list.readlines()
num_words = len(words)
logging.info("Loaded {} words".format(num_words))

url_base = "http://"
if args.ssl:
    url_base = "https://"
url_base = url_base + args.host + "/~"
logging.info("Base URL: {}".format(url_base))

class AtomicCounter:
    """An atomic, thread-safe incrementing counter.
    >>> counter = AtomicCounter()
    >>> counter.increment()
    1
    >>> counter.increment(4)
    5
    >>> counter = AtomicCounter(42.5)
    >>> counter.value
    42.5
    >>> counter.increment(0.5)
    43.0
    >>> counter = AtomicCounter()
    >>> def incrementor():
    ...     for i in range(100000):
    ...         counter.increment()
    >>> threads = []
    >>> for i in range(4):
    ...     thread = threading.Thread(target=incrementor)
    ...     thread.start()
    ...     threads.append(thread)
    >>> for thread in threads:
    ...     thread.join()
    >>> counter.value
    400000
    """
    def __init__(self, initial=0):
        """Initialize a new atomic counter to given initial value (default 0)."""
        self.value = initial
        self._lock = threading.Lock()

    def increment(self, num=1):
        """Atomically increment the counter by num (default 1) and return the
        new value.
        """
        with self._lock:
            self.value += num
            return self.value

progress = AtomicCounter()

# Print iterations progress
def printProgressBar (total, prefix = '', suffix = '', decimals = 1, length = 100, fill = '█'):
    """
    Call in a loop to create terminal progress bar
    @params:
        total       - Required  : total iterations (Int)
        prefix      - Optional  : prefix string (Str)
        suffix      - Optional  : suffix string (Str)
        decimals    - Optional  : positive number of decimals in percent complete (Int)
        length      - Optional  : character length of bar (Int)
        fill        - Optional  : bar fill character (Str)
    """
    iteration = progress.value
    percent = ("{0:." + str(decimals) + "f}").format(100 * (iteration / float(total)))
    filledLength = int(length * iteration // total)
    bar = fill * filledLength + '-' * (length - filledLength)
    print('\r%s |%s| %s%% %s' % (prefix, bar, percent, suffix), end = '\r', file=sys.stderr)
    # Print New Line on Complete
    if iteration == total: 
        print()

def test_url(word, timeout):
    word = word.strip()
    plain_word = unquote(word)
    logging.debug("Checking {}".format(url_base + plain_word))
    with requests.get(url_base + word, timeout=args.timeout) as req:
        if req.status_code != args.error_code:
            print("\n\r{} exists on {}".format(plain_word, args.host))
    progress.increment()
    printProgressBar(num_words, prefix = 'Progress:', suffix = 'Complete', length = 50)

futures = []
printProgressBar(num_words, prefix = 'Progress:', suffix = 'Complete', length = 50)
with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as pool:
    futures = [pool.submit(test_url, word, args.timeout) for word in words]

print("\r\n", file=sys.stderr)