-
-
Save joeleonjr/98b5f3b629a049954ed7bac67a80451f to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import aiohttp | |
import asyncio | |
import csv | |
import ssl | |
# Ignore SSL errors for HTTPS requests | |
ssl_context = ssl.create_default_context() | |
ssl_context.check_hostname = False | |
ssl_context.verify_mode = ssl.CERT_NONE | |
async def check_git_config(url, session): | |
try: | |
async with session.get(url, ssl=ssl_context, timeout=10) as response: | |
if response.status == 200: | |
content = await response.text() | |
return content.startswith("[core]") | |
except (aiohttp.ClientError, asyncio.TimeoutError): | |
pass | |
return False | |
async def process_domain(domain): | |
async with aiohttp.ClientSession() as session: | |
http_result = await check_git_config(f'http://{domain}/.git/config', session) | |
https_result = await check_git_config(f'https://{domain}/.git/config', session) | |
if http_result or https_result: | |
print(f"{domain} has a valid .git/config") | |
async def main(): | |
# Limit the number of concurrent requests | |
sem = asyncio.Semaphore(50) | |
async def process_domain_with_limit(domain): | |
async with sem: | |
await process_domain(domain) | |
tasks = [] | |
with open("top-1m.csv", newline="") as csvfile: | |
csv_reader = csv.reader(csvfile) | |
for row in csv_reader: | |
domain = row[1] | |
task = asyncio.ensure_future(process_domain_with_limit(domain)) | |
tasks.append(task) | |
await asyncio.gather(*tasks) | |
if __name__ == "__main__": | |
loop = asyncio.get_event_loop() | |
loop.run_until_complete(main()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment