408 lines
12 KiB
Python
408 lines
12 KiB
Python
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", True):
|
|
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><big>)
|
|
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", True):
|
|
customLinksJob()
|
|
|
|
|
|
# if os.getenv("PROXMOX_REPOS", True):
|
|
# proxmoxJob()
|
|
|
|
|
|
# if os.getenv("DOCKER_REPOS", True):
|
|
# dockerJob()
|
|
|
|
|
|
# if os.getenv("DEBIAN_REPOS", True):
|
|
# debianJob()
|
|
|
|
|
|
# if os.getenv("OPNSENSE_REPOS", True):
|
|
# opnsenseJob() |