From 3c17b0a9ab68d6ab6db64278ffc75fa7fb72aacf Mon Sep 17 00:00:00 2001 From: YuruC3 Date: Sat, 24 Jan 2026 22:53:27 +0100 Subject: [PATCH] Consolidated all functions into one --- AiO_Container/code/main.py | 408 +++++++++++++++++++++++++++++++++++++ 1 file changed, 408 insertions(+) create mode 100644 AiO_Container/code/main.py diff --git a/AiO_Container/code/main.py b/AiO_Container/code/main.py new file mode 100644 index 0000000..9612f13 --- /dev/null +++ b/AiO_Container/code/main.py @@ -0,0 +1,408 @@ +import requests, schedule, time, os +from bs4 import BeautifulSoup +from whatDomain import * +# import whatDomain + +# Debian +DEBMIRRORURL = str(os.getenv("DEBMIRRORURL", "https://www.debian.org/mirror/list")) +DEBSECURITYURL = str(os.getenv("DEBSECURITYURL", "https://security.debian.org/debian-security/")) +DEBIANMIRRORLISTV4 = "/etc/debmirror/DebianMirrorListV4" +DEBIANMIRRORLISTV6 = "/etc/debmirror/DebianMirrorListV6" + +target_countries = set([ + # Europe + "Austria", "Belgium", "Bulgaria", "Croatia", "Czech Republic", "Denmark", + "Estonia", "Finland", "France", "Germany", "Greece", "Hungary", "Iceland", + "Ireland", "Italy", "Latvia", "Lithuania", "Netherlands", "Norway", "Poland", + "Portugal", "Romania", "Slovakia", "Slovenia", "Spain", "Sweden", "Switzerland", + "United Kingdom", "Moldova", + # America + "Argentina", "Brazil", "Canada", "Chile", "Colombia", "Costa Rica", "Ecuador", + "Mexico", "Peru", "United States", "Uruguay", "Venezuela" #, + # Others + # "Security", "Extras" +]) + + +# Proxmox +PROXURL = ["http://download.proxmox.com/debian/", "https://enterprise.proxmox.com/debian/pve/",] +PROXMOXMIRRORLISTV4 = "/etc/debmirror/ProxmoxMirrorListV4" +PROXMOXMIRRORLISTV6 = "/etc/debmirror/ProxmoxMirrorListV6" + + +# Docker +DOCKERURLS = ["https://download.docker.com/linux/debian/", + "https://nvidia.github.io/libnvidia-container/stable/deb/", + "https://nvidia.github.io/libnvidia-container/experimental/deb/"] +DOCKERMIRRORLISTV4 = "/etc/debmirror/DockerMirrorListV4" +DOCKERMIRRORLISTV6 = "/etc/debmirror/DockerMirrorListV6" + + +# OPNsense +OPNSNSMIRRORURL = str(os.getenv("OPNSNSMIRRORURL", "https://opnsense.org/download/#full-mirror-listing")) +OPNSENSEMIRRORLISTV4 = "/etc/debmirror/OPNsenseMirrorListV4" +OPNSENSEMIRRORLISTV6 = "/etc/debmirror/OPNsenseMirrorListV6" + + +# Custom +EXTRAURL = [] +if os.getenv("EXTRA_REPOS", 1): + try: + with open(os.getenv("REPOFILE", "customRepoList.list"), 'r') as repoListFile: + for repoUrl in repoListFile: + # print(repoUrl.strip()) + EXTRAURL.append(repoUrl.strip()) + except FileNotFoundError: + print("File not passed") + exit(2) + +CUSTOMMIRRORLISTV4 = "/etc/debmirror/CustomMirrorListV4" +CUSTOMMIRRORLISTV6 = "/etc/debmirror/CustomMirrorListV6" + + +# Helper +def sanitizeURL(inpurl: str): + if "https://" in inpurl: + outurl = inpurl[8:] + elif "http://" in inpurl: + outurl = inpurl[7:] + # elif " " in inpurl: + # return 7 + elif "http://" or "https://" not in inpurl: + outurl = inpurl + else: + return inpurl + + i = 0 + for char in outurl: + i += 1 + if char == "/": + outurl = outurl[:i] + + if char == "/": + outurl = outurl[:-1] + return outurl + + + +# Debian +def sanitizeUrlsGodWhatTheFuckIsThis(SoupInput: BeautifulSoup): + + outMirrorDict = {} + current_country = None + + # Iterate through all table rows + for table in SoupInput.find_all("table"): + for row in table.find_all("tr"): + # Check for country name in a full-row header () + strong = row.find("strong") + if strong: + country_name = strong.get_text(strip=True) + if country_name in target_countries: + current_country = country_name + else: + current_country = None + continue # move to next row + + # Check for inline country name in first column + cols = row.find_all("td") + if len(cols) >= 2: + possible_country = cols[0].get_text(strip=True) + link_tag = cols[1].find("a", href=True) + if possible_country in target_countries: + current_country = possible_country + if current_country and link_tag: + url = link_tag['href'] + if current_country not in outMirrorDict: + outMirrorDict[current_country] = [] + outMirrorDict[current_country].append(url) + + # if os.environ["SECURITYREPOS"]: + outMirrorDict.update({"Security": DEBSECURITYURL}) + + return outMirrorDict + +def debianJob(): + + print("\nStarting lookup for Debian") + + payload = requests.get(DEBMIRRORURL) + LeSoup = BeautifulSoup(payload.content, "html.parser") + + LeMirrorDict = sanitizeUrlsGodWhatTheFuckIsThis(LeSoup) + + # print(LeMirrorDict) + + with open(DEBIANMIRRORLISTV4, "r",) as fR, open(DEBIANMIRRORLISTV4, "w",) as fW: + + for key, urls in LeMirrorDict.items(): + # print(urls) + if key in target_countries: + + for url in urls: + # print(url) + + if url not in fR: + + goodurl = sanitizeURL(url) + + if goodurl == 7: + continue + # print(goodurl) + + ip4Dict = ermWhatATheIpFromDomainYaCrazy(goodurl) + + try: + for key, ip in ip4Dict.items(): + print(ip) + + fW.write(ip + "/32" + "\n") + except AttributeError: + continue + + + with open(DEBIANMIRRORLISTV6, "r",) as fR, open(DEBIANMIRRORLISTV6, "w",) as fW: + + for key, urls in LeMirrorDict.items(): + if key in target_countries: + + for url in urls: + + if url not in fR: + + goodurl = sanitizeURL(url) + + if goodurl == 7: + continue + # print(goodurl) + + ip6Dict = ermWhatAAAATheIpFromDomainYaCrazy(goodurl) + + try: + for key, ip in ip6Dict.items(): + # print(ip) + + fW.write(ip + "/128" + "\n") + except AttributeError: + continue + + + +# Proxmox +def proxmoxJob(): + + print("\nStarting lookup for Proxmox") + + with open(PROXMOXMIRRORLISTV4, "r",) as fR, open(PROXMOXMIRRORLISTV4, "w",) as fW: + + for url in PROXURL: + if url not in fR: + sanitizedURL = sanitizeURL(url) + + if sanitizedURL == 7: + continue + + ip4Dict = ermWhatATheIpFromDomainYaCrazy(sanitizedURL) + + try: + for key, ip in ip4Dict.items(): + print(ip) + fW.write(ip + "/32" + "\n") + except AttributeError: + continue + + with open(PROXMOXMIRRORLISTV6, "r",) as fR, open(PROXMOXMIRRORLISTV6, "w",) as fW: + + for url in PROXURL: + if url not in fR: + sanitizedURL = sanitizeURL(url) + + if sanitizedURL == 7: + continue + + ip4Dict = ermWhatAAAATheIpFromDomainYaCrazy(sanitizedURL) + + try: + for key, ip in ip4Dict.items(): + print(ip) + fW.write(ip + "/128" + "\n") + except AttributeError: + continue + + +# OPNsense +def opnsenseJob(): + + print("\nStarting lookup for OPNSense") + + payload = requests.get(OPNSNSMIRRORURL) + LeOPNSoup = BeautifulSoup(payload.content, "html.parser") + + # print(LeMirrorDict) + + with open(OPNSENSEMIRRORLISTV4, "r",) as fR, open(OPNSENSEMIRRORLISTV4, "w",) as fW: + + for data in LeOPNSoup.find_all('div', class_='download_section'): + for a in data.find_all('a', href=True): + + url = a['href'] + + saniturl = sanitizeURL(url) + + if saniturl == 7: + continue + + # print(saniturl) + IPv4Dict = ermWhatATheIpFromDomainYaCrazy(saniturl) + + # print(IPv4Dict) + try: + for key, ip in IPv4Dict.items(): + print(f"Found the ipv4: {ip}") + fW.write(ip + "/32" + "\n") + + # If int returned from WhatDomain then it is error. + # Error type is described via prints from whatdomain functions + except AttributeError: + continue + + with open(OPNSENSEMIRRORLISTV6, "r",) as fR, open(OPNSENSEMIRRORLISTV6, "w",) as fW: + + for data in LeOPNSoup.find_all('div', class_='download_section'): + for a in data.find_all('a', href=True): + + url = a['href'] + + saniturl = sanitizeURL(url) + + if saniturl == 7: + continue + + # print(saniturl) + IPv6Dict = ermWhatAAAATheIpFromDomainYaCrazy(saniturl) + + # print(IPv6Dict) + try: + for key, ip in IPv6Dict.items(): + print(f"Found the ipv6: {ip}") + fW.write(ip + "/128" + "\n") + + # If int returned from WhatDomain then it is error. + # Error type is described via prints from whatdomain functions + except AttributeError: + continue + +# Docker +def dockerJob(): + + print("\nStarting lookup for Docker") + + with open(DOCKERMIRRORLISTV4, "r",) as fR, open(DOCKERMIRRORLISTV4, "w",) as fW: + + for url in DOCKERURLS: + if url not in fR: + sanitizedURL = sanitizeURL(url) + + if sanitizedURL == 7: + continue + + ip4Dict = ermWhatATheIpFromDomainYaCrazy(sanitizedURL) + + try: + for key, ip in ip4Dict.items(): + print(ip) + fW.write(ip + "/32" + "\n") + except AttributeError: + continue + with open(DOCKERMIRRORLISTV6, "r",) as fR, open(DOCKERMIRRORLISTV6, "w",) as fW: + + for url in DOCKERURLS: + if url not in fR: + sanitizedURL = sanitizeURL(url) + + if sanitizedURL == 7: + continue + + ip4Dict = ermWhatAAAATheIpFromDomainYaCrazy(sanitizedURL) + + try: + for key, ip in ip4Dict.items(): + print(ip) + fW.write(ip + "/128" + "\n") + except AttributeError: + continue + +# Custom +def customLinksJob(): + + print("\nStarting lookup for Custom domains") + + with open(CUSTOMMIRRORLISTV4, "w",) as fW, open(CUSTOMMIRRORLISTV4, "r",) as fR: + + for url in EXTRAURL: + # print(url) + goodurl = sanitizeURL(url) + # print(goodurl) + if url not in fR: + if goodurl == 7: + continue + # print(goodurl) + + # ip4Dict = ermWhatATheIpFromDomainYaCrazy(url) + ip4Dict = ermWhatATheIpFromDomainYaCrazy(goodurl) + + try: + for key, ip in ip4Dict.items(): + print(ip + "/32") + + fW.write(ip + "/32" + "\n") + except AttributeError: + continue + + + with open(CUSTOMMIRRORLISTV6, "w",) as fW, open(CUSTOMMIRRORLISTV6, "r",) as fR: + + for url in EXTRAURL: + goodurl = sanitizeURL(url) + # print(goodurl) + if url not in fR: + + if goodurl == 7: + continue + # print(goodurl) + + # ip6Dict = ermWhatAAAATheIpFromDomainYaCrazy(url) + ip6Dict = ermWhatAAAATheIpFromDomainYaCrazy(goodurl) + + try: + for key, ip in ip6Dict.items(): + print(ip + "/128") + + fW.write(ip + "/128" + "\n") + except AttributeError: + continue + + + + + +if __name__ == "__main__": + + if os.getenv("EXTRA_REPOS", 1): + customLinksJob() + + + if os.getenv("PROXMOX_REPOS", 1): + proxmoxJob() + + + if os.getenv("DOCKER_REPOS", 1): + dockerJob() + + + if os.getenv("DEBIAN_REPOS", 1): + debianJob() + + + if os.getenv("OPNSENSE_REPOS", 1): + opnsenseJob() \ No newline at end of file