admin bugfix, user agent randomization, can connect clarity increase

parent 81643ead
......@@ -3,6 +3,7 @@ from datetime import datetime
import pytz
from django.contrib import admin
from django.contrib.humanize.templatetags.humanize import naturaltime
from django.urls import reverse
from django.utils.html import format_html
from jet.admin import CompactInline
......@@ -148,16 +149,16 @@ class UrlAdmin(admin.ModelAdmin):
@staticmethod
def unresolvable_for(self):
if self.not_resolvable_since:
return "%s days" % (datetime.now(pytz.utc) - self.not_resolvable_since).days
if self.not_resolvable and self.not_resolvable_since:
return naturaltime(self.not_resolvable_since)
else:
return "-"
# todo: further humanize this.
@staticmethod
def dead_for(self):
if self.is_dead_since:
return "%s days" % (datetime.now(pytz.utc) - self.is_dead_since).days
if self.is_dead and self.is_dead_since:
return naturaltime(self.is_dead_since)
else:
return "-"
......
......@@ -15,8 +15,9 @@ class Command(BaseCommand):
help = 'Development command'
def handle(self, *args, **options):
test_can_connect_to_organization()
# as a task
develop_determineratings()
# develop_determineratings()
# reset_onboard()
# rebuild_ratings()
......@@ -135,3 +136,15 @@ def develop_determineratings():
# add_organization_rating(organization, create_history=True)
# create one for NOW, not this night. This is a bug :)
# add_organization_rating(organization)
def test_can_connect_to_organization():
from failmap_admin.scanners.scanner_http import can_connect, get_ips
organization = Organization.objects.filter(name="Zederik").get()
urls = Url.objects.all().filter(organization=organization)
for url in urls:
ipv4, ipv6 = get_ips(url.url)
if ipv4:
logger.info(can_connect("http", url, 80, ipv4))
if ipv6:
logger.info(can_connect("http", url, 80, ipv6))
......@@ -135,7 +135,9 @@ class UrlIp(models.Model):
rdns_name = models.CharField(
max_length=255,
help_text="The reverse name can be a server name, containing a provider or anything else."
"It might contain the name of a yet undiscovered url or hint to a service.")
"It might contain the name of a yet undiscovered url or hint to a service.",
blank=True
)
discovered_on = models.DateTimeField(blank=True, null=True)
......
......@@ -162,6 +162,14 @@ def scan_url(protocol: str, url: Url, port: int):
resolve_task.apply_async()
def scan_url_direct(protocol: str, url: Url, port: int):
resolve_and_scan(protocol, url, port)
# todo: apply rate limiting
# also rate limited for the same reason as can_connect is rate limited.
# would this be faster the ip discovery and actual scan grow to far apart.
# also it would mean an intense series of questions to the dns server.
@app.task
def resolve_and_scan(protocol: str, url: Url, port: int):
ips = get_ips(url.url)
......@@ -180,7 +188,6 @@ def resolve_and_scan(protocol: str, url: Url, port: int):
url_revive_task = revive_url.s(url)
url_revive_task.apply_async()
# todo: switch between ipv4 and ipv6, make tasks for different workers.
(ipv4, ipv6) = ips
if ipv4:
connect_task = can_connect.s(protocol, url, port, ipv4) # Network task
......@@ -270,12 +277,15 @@ def can_connect(protocol: str, url: Url, port: int, ip: str) -> bool:
Todo: remove IP from endpoints. (change for version 1.1)
"""
if not ip:
raise ValueError("ip not supplied.")
if ":" in ip:
uri = "%s://[%s]:%s" % (protocol, ip, port)
else:
uri = "%s://%s:%s" % (protocol, ip, port)
logger.debug("Attempting connect on: %s: host: %s" % (uri, url.url))
logger.debug("Attempting connect on: %s: host: %s IP: %s" % (uri, url.url, ip))
try:
"""
......@@ -305,7 +315,8 @@ def can_connect(protocol: str, url: Url, port: int, ip: str) -> bool:
r = requests.get(uri, timeout=(30, 30),
allow_redirects=False, # redirect = connection
verify=False, # any tls = connection
headers={'Host': url.url})
headers={'Host': url.url,
'User-Agent': get_random_user_agent()})
if r.status_code:
logger.debug("%s: Host: %s Status: %s" % (uri, url.url, r.status_code))
return True
......@@ -551,3 +562,84 @@ def test_network(code_location=""):
raise ConnectionError("Could not reach IPv6 Network via %s." % code_location)
else:
logger.info("IPv6 could be reached via %s" % code_location)
def verify_is_secure(endpoint: Endpoint):
# i've seen qualys saying there is no TLS, while there is! So qualys kills the endpoint, this adds a new one.
scan_urls(['https'], [endpoint.url], [443])
# might hopefully result in a new endpoint
endpoints = Endpoint.objects.all().filter(url=endpoint.url, is_dead=False, protocol="https", port=443,
ip_version=endpoint.ip_version)
if endpoints:
logger.debug("Url does seem to be secure after all: %s" % endpoint.url)
return True
logger.debug("Url is still not secure: %s" % endpoint.url)
return False
def redirects_to_safety(endpoint: Endpoint):
"""
Also includes the ip-version of the endpoint.
:param endpoint:
:return:
"""
import requests
from requests import ReadTimeout, ConnectTimeout, HTTPError, Timeout, ConnectionError
(ipv4, ipv6) = get_ips(endpoint.url.url)
if endpoint.ip_version == 4:
uri = "%s://%s:%s" % ("http", ipv4, "80")
else:
uri = "%s://[%s]:%s" % ("http", ipv6, "80")
try:
response = requests.get(uri,
timeout=(30, 30), # allow for insane network lag
allow_redirects=True, # point is: redirects to safety
verify=False, # certificate validity is checked elsewhere, having some https > none
headers={'Host': endpoint.url.url,
'User-Agent': get_random_user_agent()})
if response.history:
logger.debug("Request was redirected, there is hope. Redirect path:")
for resp in response.history:
logger.debug("%s: %s" % (resp.status_code, resp.url))
logger.debug("Final destination:")
logger.debug("%s: %s" % (response.status_code, response.url))
if response.url.startswith("https://"):
logger.debug("Url starts with https, so it redirects to safety.")
return True
logger.debug("Url is not redirecting to a safe url.")
return False
else:
logger.debug("Request was not redirected, so not going to a safe url.")
return False
except (ConnectTimeout, HTTPError, ReadTimeout, Timeout, ConnectionError, requests.exceptions.TooManyRedirects):
logger.debug("Request resulted into an error, it's not redirecting properly.")
return False
# http://useragentstring.com/pages/useragentstring.php/
def get_random_user_agent():
user_agents = [
# Samsung Galaxy S6
"Mozilla/5.0 (Linux; Android 6.0.1; SM-G920V Build/MMB29K) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/52.0.2743.98 Mobile Safari/537.36",
# HTC One M9
"Mozilla/5.0 (Linux; Android 6.0; HTC One M9 Build/MRA58K) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/52.0.2743.98 Mobile Safari/537.36",
# Windows 10 with Edge
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/42.0.2311.135 Safari/537.36 Edge/12.246",
# Safari
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_2) AppleWebKit/601.3.9 "
"(KHTML, like Gecko) Version/9.0.2 Safari/601.3.9",
# Windows 7
"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/47.0.2526.111 Safari/537.36"
]
return random.choice(user_agents)
......@@ -18,8 +18,7 @@ from celery import group
from failmap_admin.organizations.models import Url
from failmap_admin.scanners.endpoint_scan_manager import EndpointScanManager
from failmap_admin.scanners.scanner_http import scan_urls as scanner_http_scan_urls
from failmap_admin.scanners.scanner_http import get_ips
from failmap_admin.scanners.scanner_http import redirects_to_safety, verify_is_secure
from ..celery import app
from .models import Endpoint
......@@ -130,6 +129,8 @@ def scan_url(url: Url):
logger.debug("This url seems to have no https at all: %s" % url)
logger.debug("Checking if they exist, to be sure there is nothing.")
# todo: doesn't work anymore, as it's async
# quick fix: run it again after the discovery tasks have finished.
if not verify_is_secure(http_v4_endpoint):
logger.info("Checking if the URL redirects to a secure url: %s" % url)
......@@ -157,60 +158,3 @@ def scan_url(url: Url):
scan_manager.add_scan("plain_https", http_v6_endpoint, "0", cleaned_up)
return 'done'
def verify_is_secure(endpoint: Endpoint):
# i've seen qualys saying there is no TLS, while there is! So qualys kills the endpoint, this adds a new one.
scanner_http_scan_urls(['https'], [endpoint.url], [443])
# might hopefully result in a new endpoint
endpoints = Endpoint.objects.all().filter(url=endpoint.url, is_dead=False, protocol="https", port=443,
ip_version=endpoint.ip_version)
if endpoints:
logger.debug("Url does seem to be secure after all: %s" % endpoint.url)
return True
logger.debug("Url is still not secure: %s" % endpoint.url)
return False
def redirects_to_safety(endpoint: Endpoint):
"""
Also includes the ip-version of the endpoint.
:param endpoint:
:return:
"""
import requests
from requests import ReadTimeout, ConnectTimeout, HTTPError, Timeout, ConnectionError
(ipv4, ipv6) = get_ips(endpoint.url.url)
if endpoint.ip_version == 4:
uri = "%s://%s:%s" % ("http", ipv4, "80")
else:
uri = "%s://[%s]:%s" % ("http", ipv6, "80")
try:
response = requests.get(uri,
timeout=(30, 30), # allow for insane network lag
allow_redirects=True, # point is: redirects to safety
verify=False, # certificate validity is checked elsewhere, having some https > none
headers={'Host': endpoint.url.url})
if response.history:
logger.debug("Request was redirected, there is hope. Redirect path:")
for resp in response.history:
logger.debug("%s: %s" % (resp.status_code, resp.url))
logger.debug("Final destination:")
logger.debug("%s: %s" % (response.status_code, response.url))
if response.url.startswith("https://"):
logger.debug("Url starts with https, so it redirects to safety.")
return True
logger.debug("Url is not redirecting to a safe url.")
return False
else:
logger.debug("Request was not redirected, so not going to a safe url.")
return False
except (ConnectTimeout, HTTPError, ReadTimeout, Timeout, ConnectionError, requests.exceptions.TooManyRedirects):
logger.debug("Request resulted into an error, it's not redirecting properly.")
return False
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