mypy: fix up typing
This commit is contained in:
parent
6c88a57ad5
commit
6423d93519
6 changed files with 13 additions and 10 deletions
|
@ -72,7 +72,7 @@ def describe_brn(s: str) -> ResponseReturnValue:
|
||||||
).first()
|
).first()
|
||||||
if not proxy:
|
if not proxy:
|
||||||
return s
|
return s
|
||||||
return markupsafe.Markup( # type: ignore[no-untyped-call]
|
return markupsafe.Markup(
|
||||||
f"Proxy: {proxy.url}<br>({proxy.origin.group.group_name}: {proxy.origin.domain_name})")
|
f"Proxy: {proxy.url}<br>({proxy.origin.group.group_name}: {proxy.origin.domain_name})")
|
||||||
if parts[5].startswith("quota/"):
|
if parts[5].startswith("quota/"):
|
||||||
if parts[4] == "cloudfront":
|
if parts[4] == "cloudfront":
|
||||||
|
|
|
@ -37,7 +37,7 @@ class EditOriginForm(FlaskForm): # type: ignore
|
||||||
submit = SubmitField('Save Changes')
|
submit = SubmitField('Save Changes')
|
||||||
|
|
||||||
|
|
||||||
def final_domain_name(domain_name) -> str:
|
def final_domain_name(domain_name: str) -> str:
|
||||||
session = requests.Session()
|
session = requests.Session()
|
||||||
r = session.get(f"https://{domain_name}/", allow_redirects=True, timeout=10)
|
r = session.get(f"https://{domain_name}/", allow_redirects=True, timeout=10)
|
||||||
return urllib.parse.urlparse(r.url).netloc
|
return urllib.parse.urlparse(r.url).netloc
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
import logging
|
import logging
|
||||||
from abc import abstractmethod
|
from abc import abstractmethod
|
||||||
from typing import Tuple, List, Callable
|
from typing import Tuple, List, Callable, Optional
|
||||||
|
|
||||||
from app.extensions import db
|
from app.extensions import db
|
||||||
from app.models.activity import Activity
|
from app.models.activity import Activity
|
||||||
|
@ -10,6 +10,9 @@ from app.terraform import BaseAutomation
|
||||||
|
|
||||||
|
|
||||||
class BlockBridgeAutomation(BaseAutomation):
|
class BlockBridgeAutomation(BaseAutomation):
|
||||||
|
ips: List[str]
|
||||||
|
fingerprints: List[str]
|
||||||
|
hashed_fingerprints: List[str]
|
||||||
patterns: List[str]
|
patterns: List[str]
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
|
@ -21,7 +24,7 @@ class BlockBridgeAutomation(BaseAutomation):
|
||||||
self.hashed_fingerprints = []
|
self.hashed_fingerprints = []
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|
||||||
def perform_deprecations(self, ids: List[str], bridge_select_func: Callable[[str], Bridge]
|
def perform_deprecations(self, ids: List[str], bridge_select_func: Callable[[str], Optional[Bridge]]
|
||||||
) -> List[Tuple[str, str]]:
|
) -> List[Tuple[str, str]]:
|
||||||
rotated = []
|
rotated = []
|
||||||
for id_ in ids:
|
for id_ in ids:
|
||||||
|
@ -80,7 +83,7 @@ class BlockBridgeAutomation(BaseAutomation):
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
|
||||||
def get_bridge_by_ip(ip: str) -> Bridge:
|
def get_bridge_by_ip(ip: str) -> Optional[Bridge]:
|
||||||
return Bridge.query.filter( # type: ignore[no-any-return]
|
return Bridge.query.filter( # type: ignore[no-any-return]
|
||||||
Bridge.deprecated.is_(None),
|
Bridge.deprecated.is_(None),
|
||||||
Bridge.destroyed.is_(None),
|
Bridge.destroyed.is_(None),
|
||||||
|
@ -88,7 +91,7 @@ def get_bridge_by_ip(ip: str) -> Bridge:
|
||||||
).first()
|
).first()
|
||||||
|
|
||||||
|
|
||||||
def get_bridge_by_fingerprint(fingerprint: str) -> Bridge:
|
def get_bridge_by_fingerprint(fingerprint: str) -> Optional[Bridge]:
|
||||||
return Bridge.query.filter( # type: ignore[no-any-return]
|
return Bridge.query.filter( # type: ignore[no-any-return]
|
||||||
Bridge.deprecated.is_(None),
|
Bridge.deprecated.is_(None),
|
||||||
Bridge.destroyed.is_(None),
|
Bridge.destroyed.is_(None),
|
||||||
|
@ -96,7 +99,7 @@ def get_bridge_by_fingerprint(fingerprint: str) -> Bridge:
|
||||||
).first()
|
).first()
|
||||||
|
|
||||||
|
|
||||||
def get_bridge_by_hashed_fingerprint(hashed_fingerprint: str) -> Bridge:
|
def get_bridge_by_hashed_fingerprint(hashed_fingerprint: str) -> Optional[Bridge]:
|
||||||
return Bridge.query.filter( # type: ignore[no-any-return]
|
return Bridge.query.filter( # type: ignore[no-any-return]
|
||||||
Bridge.deprecated.is_(None),
|
Bridge.deprecated.is_(None),
|
||||||
Bridge.destroyed.is_(None),
|
Bridge.destroyed.is_(None),
|
||||||
|
|
|
@ -17,7 +17,7 @@ class BlockBridgeDnscAutomation(BlockBridgeAutomation):
|
||||||
|
|
||||||
_content: BinaryIO
|
_content: BinaryIO
|
||||||
|
|
||||||
def fetch(self):
|
def fetch(self) -> None:
|
||||||
response = requests.get(DOWNLOAD_URL,
|
response = requests.get(DOWNLOAD_URL,
|
||||||
headers={"User-Agent": "Mozilla/5.0 (Windows NT x.y; Win64; x64; rv:10.0) "
|
headers={"User-Agent": "Mozilla/5.0 (Windows NT x.y; Win64; x64; rv:10.0) "
|
||||||
"Gecko/20100101 Firefox/10.0"})
|
"Gecko/20100101 Firefox/10.0"})
|
||||||
|
|
|
@ -10,7 +10,7 @@ class BlockBridgeReachabilityAutomation(BlockBridgeAutomation):
|
||||||
|
|
||||||
_lines: List[str]
|
_lines: List[str]
|
||||||
|
|
||||||
def parse(self):
|
def parse(self) -> None:
|
||||||
for line in self._lines:
|
for line in self._lines:
|
||||||
parts = line.split("\t")
|
parts = line.split("\t")
|
||||||
if isoparse(parts[2]) < (datetime.datetime.now(datetime.timezone.utc)
|
if isoparse(parts[2]) < (datetime.datetime.now(datetime.timezone.utc)
|
||||||
|
|
|
@ -159,7 +159,7 @@ class ProxyAutomation(TerraformAutomation):
|
||||||
for group in groups:
|
for group in groups:
|
||||||
self.sp_config(group)
|
self.sp_config(group)
|
||||||
|
|
||||||
def sp_trusted_prefixes(self):
|
def sp_trusted_prefixes(self) -> str:
|
||||||
return "\n".join([f"geoip2_proxy {p};" for p in all_cdn_prefixes()])
|
return "\n".join([f"geoip2_proxy {p};" for p in all_cdn_prefixes()])
|
||||||
|
|
||||||
def sp_config(self, group: Group) -> None:
|
def sp_config(self, group: Group) -> None:
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue