Commit 822ac1a4 authored by Romain Courteaud's avatar Romain Courteaud

Check all frontend access

parent 3b0183aa
# URL checker contribution
## Install development environnment
virtualenv venv
. venv/bin/activate
pip install black pyflakes
## Check the code
pyflakes *py
black -t py37 -l 79 *py
from setuptools import setup
setup(
name="urlchecker",
version="0.0.1",
long_description=__doc__,
py_modules=["urlchecker-http", "urlchecker-cli"],
include_package_data=False,
zip_safe=True,
install_requires=[
"setuptools==40.5.0",
"requests==2.20.0",
"requests_toolbelt==0.8.0",
"peewee==2.10.1",
"click==7.0",
],
entry_points={
"console_scripts": ["urlchecker=urlchecker_cli:runUrlChecker "]
},
)
import click
import sys
@click.group()
def runUrlChecker():
pass
@runUrlChecker.command("bot", short_help="Runs url checker bot.")
@click.argument("url")
def runWebBot(url):
from urlchecker_http import WebBot
click.echo("Running url checker bot")
bot = WebBot()
return bot.run(url)
if __name__ == "__main__":
sys.exit(runUrlChecker())
from urllib.parse import urlparse
import requests
import socket
import sys
import traceback
import time
__version__ = "0.0.1"
PREFERRED_TYPE = "text/html"
TIMEOUT = 2
class BotError(Exception):
pass
class WebBot:
def getUserAgent(self):
return "%s/%s (+%s)" % (
"URLCHECKER",
__version__,
"https://lab.nexedi.com/romain/url-checker",
)
def request(
self,
method,
url,
headers=None,
stream=False,
timeout=TIMEOUT,
allow_redirects=False,
**kwargs,
):
if headers is None:
headers = {}
if "Accept" not in headers:
headers["Accept"] = "%s;q=0.9,*/*;q=0.8" % PREFERRED_TYPE
if "User-Agent" not in headers:
# XXX user agent
headers["User-Agent"] = self.getUserAgent()
kwargs["stream"] = stream
kwargs["timeout"] = timeout
kwargs["allow_redirects"] = allow_redirects
args = [url]
kwargs["headers"] = headers
try:
response = requests.get(*args, **kwargs)
except requests.exceptions.SSLError:
# XXX Enter into unknown host
response = requests.models.Response()
response.status_code = 526
except requests.exceptions.ConnectionError:
response = requests.models.Response()
response.status_code = 523
except requests.exceptions.Timeout:
response = requests.models.Response()
response.status_code = 524
except requests.exceptions.TooManyRedirects:
response = requests.models.Response()
response.status_code = 520
except:
# XXX Put it in body
print(traceback.print_exception(*(sys.exc_info())))
response = requests.models.Response()
response.status_code = 0
return response
def check(self, url):
print(url)
parsed_url = urlparse(url)
# Get the list of available IPv4 frontend CDN
hostname = parsed_url.hostname
dns_info_list = socket.getaddrinfo(hostname, "http", socket.AF_INET)
ip_list = [x[4][0] for x in dns_info_list]
for ip in ip_list:
parsed_url = parsed_url._replace(netloc=ip)
ip_url = parsed_url.geturl()
response = self.request("GET", ip_url, headers={"Host": hostname})
print(ip, response.status_code)
def run(self, url):
print(time.strftime("%Y-%m-%d %H:%M:%S"))
try:
self.check(url)
except KeyboardInterrupt:
self.stop()
except:
print("Oups, error")
raise
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment