Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 71 additions & 5 deletions ovos_utils/network_utils.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,27 @@
import socket

import requests

from ovos_utils.log import LOG


def get_network_tests_config():
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This could be made internal if its only intended to be used by the methods in this module, though it shouldn't make a difference with the way Configuration is initialized

"""Get network_tests object from mycroft.configuration."""
from ovos_config import Configuration
config = Configuration()
return config.get(
"network_tests",
{
"ip_url": 'https://api.ipify.org',
"dns_primary": "1.1.1.1",
"dns_secondary": "8.8.8.8",
"web_url": "http://nmcheck.gnome.org/check_network_status.txt",
"web_url_secondary": "https://checkonline.home-assistant.io/online.txt",
"captive_portal_url": "http://nmcheck.gnome.org/check_network_status.txt",
"captive_portal_text": "NetworkManager is online"
}
)


def get_ip():
# taken from https://stackoverflow.com/a/28950776/13703283
Expand All @@ -17,27 +38,72 @@ def get_ip():


def get_external_ip():
return requests.get('https://api.ipify.org').text
cfg = get_network_tests_config()
return requests.get(cfg["ip_url"]).text


def is_connected_dns(host="1.1.1.1"):
def is_connected_dns(host=None, port=53, timeout=3):
"""Check internet connection by connecting to DNS servers
Returns:
True if internet connection can be detected
"""

if host is None:
cfg = get_network_tests_config()
return is_connected_dns(cfg["dns_primary"]) or \
is_connected_dns(cfg["dns_secondary"])

try:
# connect to the host -- tells us if the host is actually reachable
socket.create_connection((host, 53))
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(timeout)
s.connect((host, port))
return True
except OSError:
pass
return False


def is_connected_http(host="http://duckduckgo.com"):
def is_connected_http(host=None):
"""Check internet connection by connecting to some website
Returns:
True if connection attempt succeeded
"""
if host is None:
cfg = get_network_tests_config()
return is_connected_http(cfg["web_url"]) or \
is_connected_http(cfg["web_url_secondary"])

try:
status = requests.head(host).status_code
return True
except OSError:
except:
pass
return False


def is_connected():
return any((is_connected_dns(), is_connected_http()))


def check_captive_portal(host=None, expected_text=None) -> bool:
"""Returns True if a captive portal page is detected"""
captive_portal = False

if not host or not expected_text:
cfg = get_network_tests_config()
host = host or cfg["captive_portal_url"]
expected_text = expected_text or cfg["captive_portal_url_text"]

try:
# We need to check a site that doesn't use HTTPS
html_doc = requests.get(host).text
# If something different is in the html, we likely were redirected
# to the portal page.
if expected_text not in html_doc:
captive_portal = True
except Exception:
LOG.exception("Error checking for captive portal")

return captive_portal