Skip to content

Commit

Permalink
Implement --no-proxy to disable all uses proxies and proxy providers
Browse files Browse the repository at this point in the history 
This prevents a service from setting a proxy if geofenced, and also discards any manually provided proxy from `--proxy`.
  • Loading branch information
@rlaphoenix
rlaphoenix committed Jan 9, 2024
1 parent f28a6dc commit 58cb00b
Show file tree
Hide file tree
Showing 2 changed files with 80 additions and 74 deletions.
89 changes: 47 additions & 42 deletions devine/commands/dl.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@ def cli(ctx: click.Context, **kwargs: Any) -> dl:
def __init__(
self,
ctx: click.Context,
no_proxy: bool,
profile: Optional[str] = None,
proxy: Optional[str] = None,
group: Optional[str] = None,
Expand Down Expand Up @@ -192,49 +193,52 @@ def __init__(
self.vaults.load(vault_type, **vault)
self.log.info(f"Loaded {len(self.vaults)} Vaults")

with console.status("Loading Proxy Providers...", spinner="dots"):
self.proxy_providers = []
if config.proxy_providers.get("basic"):
self.proxy_providers.append(Basic(**config.proxy_providers["basic"]))
if config.proxy_providers.get("nordvpn"):
self.proxy_providers.append(NordVPN(**config.proxy_providers["nordvpn"]))
if get_binary_path("hola-proxy"):
self.proxy_providers.append(Hola())
for proxy_provider in self.proxy_providers:
self.log.info(f"Loaded {proxy_provider.__class__.__name__}: {proxy_provider}")

if proxy:
requested_provider = None
if re.match(r"^[a-z]+:.+$", proxy, re.IGNORECASE):
# requesting proxy from a specific proxy provider
requested_provider, proxy = proxy.split(":", maxsplit=1)
if re.match(r"^[a-z]{2}(?:\d+)?$", proxy, re.IGNORECASE):
proxy = proxy.lower()
with console.status(f"Getting a Proxy to {proxy}...", spinner="dots"):
if requested_provider:
proxy_provider = next((
x
for x in self.proxy_providers
if x.__class__.__name__.lower() == requested_provider
), None)
if not proxy_provider:
self.log.error(f"The proxy provider '{requested_provider}' was not recognised.")
sys.exit(1)
proxy_uri = proxy_provider.get_proxy(proxy)
if not proxy_uri:
self.log.error(f"The proxy provider {requested_provider} had no proxy for {proxy}")
sys.exit(1)
proxy = ctx.params["proxy"] = proxy_uri
self.log.info(f"Using {proxy_provider.__class__.__name__} Proxy: {proxy}")
else:
for proxy_provider in self.proxy_providers:
self.proxy_providers = []
if no_proxy:
ctx.params["proxy"] = None
else:
with console.status("Loading Proxy Providers...", spinner="dots"):
if config.proxy_providers.get("basic"):
self.proxy_providers.append(Basic(**config.proxy_providers["basic"]))
if config.proxy_providers.get("nordvpn"):
self.proxy_providers.append(NordVPN(**config.proxy_providers["nordvpn"]))
if get_binary_path("hola-proxy"):
self.proxy_providers.append(Hola())
for proxy_provider in self.proxy_providers:
self.log.info(f"Loaded {proxy_provider.__class__.__name__}: {proxy_provider}")

if proxy:
requested_provider = None
if re.match(r"^[a-z]+:.+$", proxy, re.IGNORECASE):
# requesting proxy from a specific proxy provider
requested_provider, proxy = proxy.split(":", maxsplit=1)
if re.match(r"^[a-z]{2}(?:\d+)?$", proxy, re.IGNORECASE):
proxy = proxy.lower()
with console.status(f"Getting a Proxy to {proxy}...", spinner="dots"):
if requested_provider:
proxy_provider = next((
x
for x in self.proxy_providers
if x.__class__.__name__.lower() == requested_provider
), None)
if not proxy_provider:
self.log.error(f"The proxy provider '{requested_provider}' was not recognised.")
sys.exit(1)
proxy_uri = proxy_provider.get_proxy(proxy)
if proxy_uri:
proxy = ctx.params["proxy"] = proxy_uri
self.log.info(f"Using {proxy_provider.__class__.__name__} Proxy: {proxy}")
break
else:
self.log.info(f"Using explicit Proxy: {proxy}")
if not proxy_uri:
self.log.error(f"The proxy provider {requested_provider} had no proxy for {proxy}")
sys.exit(1)
proxy = ctx.params["proxy"] = proxy_uri
self.log.info(f"Using {proxy_provider.__class__.__name__} Proxy: {proxy}")
else:
for proxy_provider in self.proxy_providers:
proxy_uri = proxy_provider.get_proxy(proxy)
if proxy_uri:
proxy = ctx.params["proxy"] = proxy_uri
self.log.info(f"Using {proxy_provider.__class__.__name__} Proxy: {proxy}")
break
else:
self.log.info(f"Using explicit Proxy: {proxy}")

ctx.obj = ContextData(
config=self.service_config,
Expand Down Expand Up @@ -274,6 +278,7 @@ def result(
skip_dl: bool,
export: Optional[Path],
cdm_only: Optional[bool],
no_proxy: bool,
no_folder: bool,
no_source: bool,
workers: int,
Expand Down
65 changes: 33 additions & 32 deletions devine/core/service.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,39 +41,40 @@ def __init__(self, ctx: click.Context):
self.session = self.get_session()
self.cache = Cacher(self.__class__.__name__)

if ctx.parent:
self.proxy = ctx.parent.params["proxy"]
else:
self.proxy = None

if not self.proxy:
# don't override the explicit proxy set by the user, even if they may be geoblocked
with console.status("Checking if current region is Geoblocked...", spinner="dots"):
if self.GEOFENCE:
# no explicit proxy, let's get one to GEOFENCE if needed
current_region = get_ip_info(self.session)["country"].lower()
if any(x.lower() == current_region for x in self.GEOFENCE):
self.log.info("Service is not Geoblocked in your region")
if not ctx.parent or not ctx.parent.params.get("no_proxy"):
if ctx.parent:
proxy = ctx.parent.params["proxy"]
else:
proxy = None

if not proxy:
# don't override the explicit proxy set by the user, even if they may be geoblocked
with console.status("Checking if current region is Geoblocked...", spinner="dots"):
if self.GEOFENCE:
# no explicit proxy, let's get one to GEOFENCE if needed
current_region = get_ip_info(self.session)["country"].lower()
if any(x.lower() == current_region for x in self.GEOFENCE):
self.log.info("Service is not Geoblocked in your region")
else:
requested_proxy = self.GEOFENCE[0] # first is likely main region
self.log.info(f"Service is Geoblocked in your region, getting a Proxy to {requested_proxy}")
for proxy_provider in ctx.obj.proxy_providers:
proxy = proxy_provider.get_proxy(requested_proxy)
if proxy:
self.log.info(f"Got Proxy from {proxy_provider.__class__.__name__}")
break
else:
requested_proxy = self.GEOFENCE[0] # first is likely main region
self.log.info(f"Service is Geoblocked in your region, getting a Proxy to {requested_proxy}")
for proxy_provider in ctx.obj.proxy_providers:
self.proxy = proxy_provider.get_proxy(requested_proxy)
if self.proxy:
self.log.info(f"Got Proxy from {proxy_provider.__class__.__name__}")
break
else:
self.log.info("Service has no Geofence")

if self.proxy:
self.session.proxies.update({"all": self.proxy})
proxy_parse = urlparse(self.proxy)
if proxy_parse.username and proxy_parse.password:
self.session.headers.update({
"Proxy-Authorization": base64.b64encode(
f"{proxy_parse.username}:{proxy_parse.password}".encode("utf8")
).decode()
})
self.log.info("Service has no Geofence")

if proxy:
self.session.proxies.update({"all": proxy})
proxy_parse = urlparse(proxy)
if proxy_parse.username and proxy_parse.password:
self.session.headers.update({
"Proxy-Authorization": base64.b64encode(
f"{proxy_parse.username}:{proxy_parse.password}".encode("utf8")
).decode()
})

# Optional Abstract functions
# The following functions may be implemented by the Service.
Expand Down

0 comments on commit 58cb00b

Please sign in to comment.