84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
# pylint: disable=too-few-public-methods
|
|
|
|
import builtins
|
|
from datetime import datetime
|
|
from typing import List, Dict, Union, Any, Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.models.base import Pool
|
|
from app.models.mirrors import Origin, Proxy
|
|
|
|
|
|
class BC2Alternative(BaseModel):
|
|
proto: str
|
|
type: str
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
url: str
|
|
|
|
|
|
class BC2Site(BaseModel):
|
|
main_domain: str = Field(description="The main domain name of the website, excluding \"www.\" if present.",
|
|
examples=["bbc.co.uk", "bbc.com", "guardianproject.info"])
|
|
available_alternatives: List[BC2Alternative]
|
|
|
|
|
|
class BypassCensorship2(BaseModel):
|
|
version: str = Field(description="Version number of the Bypass Censorship Extension schema in use", )
|
|
sites: List[BC2Site]
|
|
|
|
class Config:
|
|
title = "Bypass Censorship Version 2"
|
|
|
|
|
|
def onion_alternative(origin: Origin) -> List[Dict[str, Any]]:
|
|
url: Optional[str] = origin.onion()
|
|
if url is None:
|
|
return []
|
|
return [{
|
|
"proto": "tor",
|
|
"type": "eotk",
|
|
"created_at": str(origin.added),
|
|
"updated_at": str(origin.updated),
|
|
"url": url}
|
|
]
|
|
|
|
|
|
def proxy_alternative(proxy: Proxy) -> Dict[str, Any]:
|
|
return {
|
|
"proto": "https",
|
|
"type": "mirror",
|
|
"created_at": str(proxy.added),
|
|
"updated_at": str(proxy.updated),
|
|
"url": proxy.url
|
|
}
|
|
|
|
|
|
def main_domain(origin: Origin) -> str:
|
|
# Both description and domain_name are required to be not null in the database schema
|
|
description: str = origin.description
|
|
if description.startswith("proxy:"):
|
|
return description[len("proxy:"):].replace("www.", "")
|
|
domain_name: str = origin.domain_name
|
|
return domain_name.replace("www.", "")
|
|
|
|
|
|
def active_proxies(origin: Origin, pool: Pool) -> List[Proxy]:
|
|
def _filter_fn(proxy: Proxy) -> bool:
|
|
return proxy.url is not None and not proxy.deprecated and not proxy.destroyed and proxy.pool_id == pool.id
|
|
return list(filter(_filter_fn, origin.proxies))
|
|
|
|
|
|
def mirror_sites(pool: Pool) -> Dict[
|
|
str, Union[str, List[Dict[str, Union[str, List[Dict[str, str]]]]]]]:
|
|
return {"version": "2.0", "sites": [{"main_domain": main_domain(origin),
|
|
"available_alternatives": onion_alternative(origin) + [
|
|
proxy_alternative(proxy) for proxy in
|
|
active_proxies(origin, pool)]} for origin in
|
|
Origin.query.order_by(Origin.domain_name).all() if
|
|
origin.destroyed is None]}
|
|
|
|
|
|
if getattr(builtins, "__sphinx_build__", False):
|
|
schema = BypassCensorship2.schema_json()
|