stevazi/addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

aggiunto check connessione e DNS

Browse Source
This commit is contained in:
mac12m99
2019-08-30 18:03:45 +02:00
parent d789aec06d
commit aa3291eedb
7 changed files with 5408 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2231
lib/httplib2/__init__.py Normal file
View File

File diff suppressed because it is too large Load Diff

2196
lib/httplib2/cacerts.txt Normal file
View File

File diff suppressed because it is too large Load Diff

42
lib/httplib2/certs.py Normal file
View File

@@ -0,0 +1,42 @@
"""Utilities for certificate management."""
import os
certifi_available = False
certifi_where = None
try:
from certifi import where as certifi_where
certifi_available = True
except ImportError:
pass
custom_ca_locater_available = False
custom_ca_locater_where = None
try:
from ca_certs_locater import get as custom_ca_locater_where
custom_ca_locater_available = True
except ImportError:
pass
BUILTIN_CA_CERTS = os.path.join(
os.path.dirname(os.path.abspath(__file__)), "cacerts.txt"
)
def where():
env = os.environ.get("HTTPLIB2_CA_CERTS")
if env is not None:
if os.path.isfile(env):
return env
else:
raise RuntimeError("Environment variable HTTPLIB2_CA_CERTS not a valid file")
if custom_ca_locater_available:
return custom_ca_locater_where()
if certifi_available:
return certifi_where()
return BUILTIN_CA_CERTS
if __name__ == "__main__":
print(where())

123
lib/httplib2/iri2uri.py Normal file
View File

@@ -0,0 +1,123 @@
"""Converts an IRI to a URI."""
__author__ = "Joe Gregorio (joe@bitworking.org)"
__copyright__ = "Copyright 2006, Joe Gregorio"
__contributors__ = []
__version__ = "1.0.0"
__license__ = "MIT"
import urlparse
# Convert an IRI to a URI following the rules in RFC 3987
#
# The characters we need to enocde and escape are defined in the spec:
#
# iprivate = %xE000-F8FF / %xF0000-FFFFD / %x100000-10FFFD
# ucschar = %xA0-D7FF / %xF900-FDCF / %xFDF0-FFEF
# / %x10000-1FFFD / %x20000-2FFFD / %x30000-3FFFD
# / %x40000-4FFFD / %x50000-5FFFD / %x60000-6FFFD
# / %x70000-7FFFD / %x80000-8FFFD / %x90000-9FFFD
# / %xA0000-AFFFD / %xB0000-BFFFD / %xC0000-CFFFD
# / %xD0000-DFFFD / %xE1000-EFFFD
escape_range = [
(0xA0, 0xD7FF),
(0xE000, 0xF8FF),
(0xF900, 0xFDCF),
(0xFDF0, 0xFFEF),
(0x10000, 0x1FFFD),
(0x20000, 0x2FFFD),
(0x30000, 0x3FFFD),
(0x40000, 0x4FFFD),
(0x50000, 0x5FFFD),
(0x60000, 0x6FFFD),
(0x70000, 0x7FFFD),
(0x80000, 0x8FFFD),
(0x90000, 0x9FFFD),
(0xA0000, 0xAFFFD),
(0xB0000, 0xBFFFD),
(0xC0000, 0xCFFFD),
(0xD0000, 0xDFFFD),
(0xE1000, 0xEFFFD),
(0xF0000, 0xFFFFD),
(0x100000, 0x10FFFD),
]
def encode(c):
retval = c
i = ord(c)
for low, high in escape_range:
if i < low:
break
if i >= low and i <= high:
retval = "".join(["%%%2X" % ord(o) for o in c.encode("utf-8")])
break
return retval
def iri2uri(uri):
"""Convert an IRI to a URI. Note that IRIs must be
passed in a unicode strings. That is, do not utf-8 encode
the IRI before passing it into the function."""
if isinstance(uri, unicode):
(scheme, authority, path, query, fragment) = urlparse.urlsplit(uri)
authority = authority.encode("idna")
# For each character in 'ucschar' or 'iprivate'
# 1. encode as utf-8
# 2. then %-encode each octet of that utf-8
uri = urlparse.urlunsplit((scheme, authority, path, query, fragment))
uri = "".join([encode(c) for c in uri])
return uri
if __name__ == "__main__":
import unittest
class Test(unittest.TestCase):
def test_uris(self):
"""Test that URIs are invariant under the transformation."""
invariant = [
u"ftp://ftp.is.co.za/rfc/rfc1808.txt",
u"http://www.ietf.org/rfc/rfc2396.txt",
u"ldap://[2001:db8::7]/c=GB?objectClass?one",
u"mailto:John.Doe@example.com",
u"news:comp.infosystems.www.servers.unix",
u"tel:+1-816-555-1212",
u"telnet://192.0.2.16:80/",
u"urn:oasis:names:specification:docbook:dtd:xml:4.1.2",
]
for uri in invariant:
self.assertEqual(uri, iri2uri(uri))
def test_iri(self):
"""Test that the right type of escaping is done for each part of the URI."""
self.assertEqual(
"http://xn--o3h.com/%E2%98%84",
iri2uri(u"http://\N{COMET}.com/\N{COMET}"),
)
self.assertEqual(
"http://bitworking.org/?fred=%E2%98%84",
iri2uri(u"http://bitworking.org/?fred=\N{COMET}"),
)
self.assertEqual(
"http://bitworking.org/#%E2%98%84",
iri2uri(u"http://bitworking.org/#\N{COMET}"),
)
self.assertEqual("#%E2%98%84", iri2uri(u"#\N{COMET}"))
self.assertEqual(
"/fred?bar=%E2%98%9A#%E2%98%84",
iri2uri(u"/fred?bar=\N{BLACK LEFT POINTING INDEX}#\N{COMET}"),
)
self.assertEqual(
"/fred?bar=%E2%98%9A#%E2%98%84",
iri2uri(iri2uri(u"/fred?bar=\N{BLACK LEFT POINTING INDEX}#\N{COMET}")),
)
self.assertNotEqual(
"/fred?bar=%E2%98%9A#%E2%98%84",
iri2uri(
u"/fred?bar=\N{BLACK LEFT POINTING INDEX}#\N{COMET}".encode("utf-8")
),
)
unittest.main()

510
lib/httplib2/socks.py Normal file
View File

@@ -0,0 +1,510 @@
"""SocksiPy - Python SOCKS module.
Version 1.00
Copyright 2006 Dan-Haim. All rights reserved.
Redistribution and use in source and binary forms, with or without modification,
are permitted provided that the following conditions are met:
1. Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
3. Neither the name of Dan Haim nor the names of his contributors may be used
to endorse or promote products derived from this software without specific
prior written permission.
THIS SOFTWARE IS PROVIDED BY DAN HAIM "AS IS" AND ANY EXPRESS OR IMPLIED
WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
EVENT SHALL DAN HAIM OR HIS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA
OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMANGE.
This module provides a standard socket-like interface for Python
for tunneling connections through SOCKS proxies.
Minor modifications made by Christopher Gilbert (http://motomastyle.com/) for
use in PyLoris (http://pyloris.sourceforge.net/).
Minor modifications made by Mario Vilas (http://breakingcode.wordpress.com/)
mainly to merge bug fixes found in Sourceforge.
"""
import base64
import socket
import struct
import sys
if getattr(socket, "socket", None) is None:
raise ImportError("socket.socket missing, proxy support unusable")
PROXY_TYPE_SOCKS4 = 1
PROXY_TYPE_SOCKS5 = 2
PROXY_TYPE_HTTP = 3
PROXY_TYPE_HTTP_NO_TUNNEL = 4
_defaultproxy = None
_orgsocket = socket.socket
class ProxyError(Exception):
pass
class GeneralProxyError(ProxyError):
pass
class Socks5AuthError(ProxyError):
pass
class Socks5Error(ProxyError):
pass
class Socks4Error(ProxyError):
pass
class HTTPError(ProxyError):
pass
_generalerrors = (
"success",
"invalid data",
"not connected",
"not available",
"bad proxy type",
"bad input",
)
_socks5errors = (
"succeeded",
"general SOCKS server failure",
"connection not allowed by ruleset",
"Network unreachable",
"Host unreachable",
"Connection refused",
"TTL expired",
"Command not supported",
"Address type not supported",
"Unknown error",
)
_socks5autherrors = (
"succeeded",
"authentication is required",
"all offered authentication methods were rejected",
"unknown username or invalid password",
"unknown error",
)
_socks4errors = (
"request granted",
"request rejected or failed",
"request rejected because SOCKS server cannot connect to identd on the client",
"request rejected because the client program and identd report different "
"user-ids",
"unknown error",
)
def setdefaultproxy(
proxytype=None, addr=None, port=None, rdns=True, username=None, password=None
):
"""setdefaultproxy(proxytype, addr[, port[, rdns[, username[, password]]]])
Sets a default proxy which all further socksocket objects will use,
unless explicitly changed.
"""
global _defaultproxy
_defaultproxy = (proxytype, addr, port, rdns, username, password)
def wrapmodule(module):
"""wrapmodule(module)
Attempts to replace a module's socket library with a SOCKS socket. Must set
a default proxy using setdefaultproxy(...) first.
This will only work on modules that import socket directly into the
namespace;
most of the Python Standard Library falls into this category.
"""
if _defaultproxy != None:
module.socket.socket = socksocket
else:
raise GeneralProxyError((4, "no proxy specified"))
class socksocket(socket.socket):
"""socksocket([family[, type[, proto]]]) -> socket object
Open a SOCKS enabled socket. The parameters are the same as
those of the standard socket init. In order for SOCKS to work,
you must specify family=AF_INET, type=SOCK_STREAM and proto=0.
"""
def __init__(
self, family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0, _sock=None
):
_orgsocket.__init__(self, family, type, proto, _sock)
if _defaultproxy != None:
self.__proxy = _defaultproxy
else:
self.__proxy = (None, None, None, None, None, None)
self.__proxysockname = None
self.__proxypeername = None
self.__httptunnel = True
def __recvall(self, count):
"""__recvall(count) -> data
Receive EXACTLY the number of bytes requested from the socket.
Blocks until the required number of bytes have been received.
"""
data = self.recv(count)
while len(data) < count:
d = self.recv(count - len(data))
if not d:
raise GeneralProxyError((0, "connection closed unexpectedly"))
data = data + d
return data
def sendall(self, content, *args):
""" override socket.socket.sendall method to rewrite the header
for non-tunneling proxies if needed
"""
if not self.__httptunnel:
content = self.__rewriteproxy(content)
return super(socksocket, self).sendall(content, *args)
def __rewriteproxy(self, header):
""" rewrite HTTP request headers to support non-tunneling proxies
(i.e. those which do not support the CONNECT method).
This only works for HTTP (not HTTPS) since HTTPS requires tunneling.
"""
host, endpt = None, None
hdrs = header.split("\r\n")
for hdr in hdrs:
if hdr.lower().startswith("host:"):
host = hdr
elif hdr.lower().startswith("get") or hdr.lower().startswith("post"):
endpt = hdr
if host and endpt:
hdrs.remove(host)
hdrs.remove(endpt)
host = host.split(" ")[1]
endpt = endpt.split(" ")
if self.__proxy[4] != None and self.__proxy[5] != None:
hdrs.insert(0, self.__getauthheader())
hdrs.insert(0, "Host: %s" % host)
hdrs.insert(0, "%s http://%s%s %s" % (endpt[0], host, endpt[1], endpt[2]))
return "\r\n".join(hdrs)
def __getauthheader(self):
auth = self.__proxy[4] + ":" + self.__proxy[5]
return "Proxy-Authorization: Basic " + base64.b64encode(auth)
def setproxy(
self,
proxytype=None,
addr=None,
port=None,
rdns=True,
username=None,
password=None,
headers=None,
):
"""setproxy(proxytype, addr[, port[, rdns[, username[, password]]]])
Sets the proxy to be used.
proxytype - The type of the proxy to be used. Three types
are supported: PROXY_TYPE_SOCKS4 (including socks4a),
PROXY_TYPE_SOCKS5 and PROXY_TYPE_HTTP
addr - The address of the server (IP or DNS).
port - The port of the server. Defaults to 1080 for SOCKS
servers and 8080 for HTTP proxy servers.
rdns - Should DNS queries be preformed on the remote side
(rather than the local side). The default is True.
Note: This has no effect with SOCKS4 servers.
username - Username to authenticate with to the server.
The default is no authentication.
password - Password to authenticate with to the server.
Only relevant when username is also provided.
headers - Additional or modified headers for the proxy connect
request.
"""
self.__proxy = (proxytype, addr, port, rdns, username, password, headers)
def __negotiatesocks5(self, destaddr, destport):
"""__negotiatesocks5(self,destaddr,destport)
Negotiates a connection through a SOCKS5 server.
"""
# First we'll send the authentication packages we support.
if (self.__proxy[4] != None) and (self.__proxy[5] != None):
# The username/password details were supplied to the
# setproxy method so we support the USERNAME/PASSWORD
# authentication (in addition to the standard none).
self.sendall(struct.pack("BBBB", 0x05, 0x02, 0x00, 0x02))
else:
# No username/password were entered, therefore we
# only support connections with no authentication.
self.sendall(struct.pack("BBB", 0x05, 0x01, 0x00))
# We'll receive the server's response to determine which
# method was selected
chosenauth = self.__recvall(2)
if chosenauth[0:1] != chr(0x05).encode():
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
# Check the chosen authentication method
if chosenauth[1:2] == chr(0x00).encode():
# No authentication is required
pass
elif chosenauth[1:2] == chr(0x02).encode():
# Okay, we need to perform a basic username/password
# authentication.
self.sendall(
chr(0x01).encode()
+ chr(len(self.__proxy[4]))
+ self.__proxy[4]
+ chr(len(self.__proxy[5]))
+ self.__proxy[5]
)
authstat = self.__recvall(2)
if authstat[0:1] != chr(0x01).encode():
# Bad response
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
if authstat[1:2] != chr(0x00).encode():
# Authentication failed
self.close()
raise Socks5AuthError((3, _socks5autherrors[3]))
# Authentication succeeded
else:
# Reaching here is always bad
self.close()
if chosenauth[1] == chr(0xFF).encode():
raise Socks5AuthError((2, _socks5autherrors[2]))
else:
raise GeneralProxyError((1, _generalerrors[1]))
# Now we can request the actual connection
req = struct.pack("BBB", 0x05, 0x01, 0x00)
# If the given destination address is an IP address, we'll
# use the IPv4 address request even if remote resolving was specified.
try:
ipaddr = socket.inet_aton(destaddr)
req = req + chr(0x01).encode() + ipaddr
except socket.error:
# Well it's not an IP number, so it's probably a DNS name.
if self.__proxy[3]:
# Resolve remotely
ipaddr = None
req = (
req
+ chr(0x03).encode()
+ chr(len(destaddr)).encode()
+ destaddr.encode()
)
else:
# Resolve locally
ipaddr = socket.inet_aton(socket.gethostbyname(destaddr))
req = req + chr(0x01).encode() + ipaddr
req = req + struct.pack(">H", destport)
self.sendall(req)
# Get the response
resp = self.__recvall(4)
if resp[0:1] != chr(0x05).encode():
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
elif resp[1:2] != chr(0x00).encode():
# Connection failed
self.close()
if ord(resp[1:2]) <= 8:
raise Socks5Error((ord(resp[1:2]), _socks5errors[ord(resp[1:2])]))
else:
raise Socks5Error((9, _socks5errors[9]))
# Get the bound address/port
elif resp[3:4] == chr(0x01).encode():
boundaddr = self.__recvall(4)
elif resp[3:4] == chr(0x03).encode():
resp = resp + self.recv(1)
boundaddr = self.__recvall(ord(resp[4:5]))
else:
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
boundport = struct.unpack(">H", self.__recvall(2))[0]
self.__proxysockname = (boundaddr, boundport)
if ipaddr != None:
self.__proxypeername = (socket.inet_ntoa(ipaddr), destport)
else:
self.__proxypeername = (destaddr, destport)
def getproxysockname(self):
"""getsockname() -> address info
Returns the bound IP address and port number at the proxy.
"""
return self.__proxysockname
def getproxypeername(self):
"""getproxypeername() -> address info
Returns the IP and port number of the proxy.
"""
return _orgsocket.getpeername(self)
def getpeername(self):
"""getpeername() -> address info
Returns the IP address and port number of the destination
machine (note: getproxypeername returns the proxy)
"""
return self.__proxypeername
def __negotiatesocks4(self, destaddr, destport):
"""__negotiatesocks4(self,destaddr,destport)
Negotiates a connection through a SOCKS4 server.
"""
# Check if the destination address provided is an IP address
rmtrslv = False
try:
ipaddr = socket.inet_aton(destaddr)
except socket.error:
# It's a DNS name. Check where it should be resolved.
if self.__proxy[3]:
ipaddr = struct.pack("BBBB", 0x00, 0x00, 0x00, 0x01)
rmtrslv = True
else:
ipaddr = socket.inet_aton(socket.gethostbyname(destaddr))
# Construct the request packet
req = struct.pack(">BBH", 0x04, 0x01, destport) + ipaddr
# The username parameter is considered userid for SOCKS4
if self.__proxy[4] != None:
req = req + self.__proxy[4]
req = req + chr(0x00).encode()
# DNS name if remote resolving is required
# NOTE: This is actually an extension to the SOCKS4 protocol
# called SOCKS4A and may not be supported in all cases.
if rmtrslv:
req = req + destaddr + chr(0x00).encode()
self.sendall(req)
# Get the response from the server
resp = self.__recvall(8)
if resp[0:1] != chr(0x00).encode():
# Bad data
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
if resp[1:2] != chr(0x5A).encode():
# Server returned an error
self.close()
if ord(resp[1:2]) in (91, 92, 93):
self.close()
raise Socks4Error((ord(resp[1:2]), _socks4errors[ord(resp[1:2]) - 90]))
else:
raise Socks4Error((94, _socks4errors[4]))
# Get the bound address/port
self.__proxysockname = (
socket.inet_ntoa(resp[4:]),
struct.unpack(">H", resp[2:4])[0],
)
if rmtrslv != None:
self.__proxypeername = (socket.inet_ntoa(ipaddr), destport)
else:
self.__proxypeername = (destaddr, destport)
def __negotiatehttp(self, destaddr, destport):
"""__negotiatehttp(self,destaddr,destport)
Negotiates a connection through an HTTP server.
"""
# If we need to resolve locally, we do this now
if not self.__proxy[3]:
addr = socket.gethostbyname(destaddr)
else:
addr = destaddr
headers = ["CONNECT ", addr, ":", str(destport), " HTTP/1.1\r\n"]
wrote_host_header = False
wrote_auth_header = False
if self.__proxy[6] != None:
for key, val in self.__proxy[6].iteritems():
headers += [key, ": ", val, "\r\n"]
wrote_host_header = key.lower() == "host"
wrote_auth_header = key.lower() == "proxy-authorization"
if not wrote_host_header:
headers += ["Host: ", destaddr, "\r\n"]
if not wrote_auth_header:
if self.__proxy[4] != None and self.__proxy[5] != None:
headers += [self.__getauthheader(), "\r\n"]
headers.append("\r\n")
self.sendall("".join(headers).encode())
# We read the response until we get the string "\r\n\r\n"
resp = self.recv(1)
while resp.find("\r\n\r\n".encode()) == -1:
resp = resp + self.recv(1)
# We just need the first line to check if the connection
# was successful
statusline = resp.splitlines()[0].split(" ".encode(), 2)
if statusline[0] not in ("HTTP/1.0".encode(), "HTTP/1.1".encode()):
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
try:
statuscode = int(statusline[1])
except ValueError:
self.close()
raise GeneralProxyError((1, _generalerrors[1]))
if statuscode != 200:
self.close()
raise HTTPError((statuscode, statusline[2]))
self.__proxysockname = ("0.0.0.0", 0)
self.__proxypeername = (addr, destport)
def connect(self, destpair):
"""connect(self, despair)
Connects to the specified destination through a proxy.
destpar - A tuple of the IP/DNS address and the port number.
(identical to socket's connect).
To select the proxy server use setproxy().
"""
# Do a minimal input check first
if (
(not type(destpair) in (list, tuple))
or (len(destpair) < 2)
or (not isinstance(destpair[0], basestring))
or (type(destpair[1]) != int)
):
raise GeneralProxyError((5, _generalerrors[5]))
if self.__proxy[0] == PROXY_TYPE_SOCKS5:
if self.__proxy[2] != None:
portnum = self.__proxy[2]
else:
portnum = 1080
_orgsocket.connect(self, (self.__proxy[1], portnum))
self.__negotiatesocks5(destpair[0], destpair[1])
elif self.__proxy[0] == PROXY_TYPE_SOCKS4:
if self.__proxy[2] != None:
portnum = self.__proxy[2]
else:
portnum = 1080
_orgsocket.connect(self, (self.__proxy[1], portnum))
self.__negotiatesocks4(destpair[0], destpair[1])
elif self.__proxy[0] == PROXY_TYPE_HTTP:
if self.__proxy[2] != None:
portnum = self.__proxy[2]
else:
portnum = 8080
_orgsocket.connect(self, (self.__proxy[1], portnum))
self.__negotiatehttp(destpair[0], destpair[1])
elif self.__proxy[0] == PROXY_TYPE_HTTP_NO_TUNNEL:
if self.__proxy[2] != None:
portnum = self.__proxy[2]
else:
portnum = 8080
_orgsocket.connect(self, (self.__proxy[1], portnum))
if destpair[1] == 443:
self.__negotiatehttp(destpair[0], destpair[1])
else:
self.__httptunnel = False
elif self.__proxy[0] == None:
_orgsocket.connect(self, (destpair[0], destpair[1]))
else:
raise GeneralProxyError((4, _generalerrors[4]))

8
platformcode/launcher.py
View File

@@ -30,6 +30,14 @@ def start():
# Test if all the required directories are created
config.verify_directories_created()
# controlla se l'utente ha qualche problema di connessione
# se lo ha: non lo fa entrare nell'addon
# se ha problemi di DNS avvia ma lascia entrare
# se tutto ok: entra nell'addon
from specials.checkhost import test_conn
import threading
threading.Thread(target=test_conn, args=(True, True, True, [], [], True)).start()
def run(item=None):
logger.info()

298
specials/checkhost.py Executable file
View File

@@ -0,0 +1,298 @@
# -*- coding: utf-8 -*-
import xbmc, xbmcgui
import xbmcaddon
import json
from platformcode import config, logger
import requests
from requests.exceptions import HTTPError
from lib import httplib2
import socket
addon = xbmcaddon.Addon()
addonname = addon.getAddonInfo('name')
addonid = addon.getAddonInfo('id')
LIST_SITE = ['https://www.google.com', 'https://www.google.it',
'http://www.ansa.it/', 'https://www.debian.org/']
# lista di siti che non verranno raggiunti con i DNS del gestore
LST_SITE_CHCK_DNS = ['https://www.italia-film.pw', 'https://casacinema.space']
class Kdicc():
def __init__(self, is_exit = True, check_dns = True, view_msg = True,
lst_urls = [], lst_site_check_dns = [], in_addon = False):
self.ip_addr = xbmc.getIPAddress()
self.dns = [xbmc.getInfoLabel('Network.DNS1Address'),
xbmc.getInfoLabel('Network.DNS2Address')]
self.check_dns = check_dns
self.is_exit = is_exit
self.lst_urls = lst_urls
self.view_msg = view_msg
self.lst_site_check_dns = lst_site_check_dns
self.urls = []
## self.in_addon = in_addon
## self.server_dns = xbmc.getInfoLabel('Network.DHCPAddress') # stringa vuota...
## self.gateway = xbmc.getInfoLabel('Network.GatewayAddress')
## self.subnet = xbmc.getInfoLabel('Network.SubnetMask')
def check_Ip(self):
"""
controllo l'ip
se ip_addr = 127.0.0.1 o ip_addr = '' allora il device non
e' connesso al modem/router
check the ip
if ip_addr = 127.0.0.1 or ip_addr = '' then the device does not
is connected to the modem/router
return: bool
"""
if self.ip_addr == '127.0.0.1' or self.ip_addr == '':
return False
else:
return True
def check_Adsl(self):
"""
controllo se il device raggiunge i siti
"""
urls = LIST_SITE
r = self.rqst(urls)
http_errr = 0
for rslt in r:
xbmc.log("check_Adsl rslt: %s" % rslt['code'], level=xbmc.LOGNOTICE)
if rslt['code'] == '111':
http_errr +=1
if len(LIST_SITE) == http_errr:
return False
else:
return True
def check_Dns(self):
"""
Controllo se i DNS raggiungono certi siti
"""
if self.lst_site_check_dns == []:
urls = LST_SITE_CHCK_DNS
else:
urls = self.lst_site_check_dns
r = self.rqst(urls)
xbmc.log("check_Dns risultato: %s" % r, level=xbmc.LOGNOTICE)
http_errr = 0
for rslt in r:
xbmc.log("check_Dns rslt: %s" % rslt['code'], level=xbmc.LOGNOTICE)
if rslt['code'] == '111':
http_errr +=1
if len(LST_SITE_CHCK_DNS) == http_errr:
return False
else:
return True
def rqst(self, lst_urls):
"""
url deve iniziare con http(s):'
return : (esito, sito, url, code, reurl)
"""
rslt_final = []
if lst_urls == []:
lst_urls = self.lst_urls
for sito in lst_urls:
rslt = {}
try:
r = requests.head(sito, allow_redirects = True)#, timeout=7) # da errore dopo l'inserimento in lib di httplib2
if r.url.endswith('/'):
r.url = r.url[:-1]
if str(sito) != str(r.url):
is_redirect = True
else:
is_redirect = False
rslt['code'] = r.status_code
rslt['url'] = str(sito)
rslt['rdrcturl'] = str(r.url)
rslt['isRedirect'] = is_redirect
rslt['history'] = r.history
xbmc.log("Risultato nel try: %s" % (r,), level=xbmc.LOGNOTICE)
except requests.exceptions.ConnectionError as conn_errr:
# Errno 10061 per s.o. win
# gli Errno 10xxx e 11xxx saranno da compattare in qualche modo?
# gli errori vengono inglobati in code = '111' in quanto in quel momento
# non vengono raggiunti per una qualsiasi causa
if '[Errno 111]' in str(conn_errr) or 'Errno 10061' in str(conn_errr) \
or 'ConnectTimeoutError' in str(conn_errr) \
or 'Errno 11002' in str(conn_errr) or 'ReadTimeout' in str(conn_errr) \
or 'Errno 11001' in str(conn_errr): # questo errore è anche nel code: -2
rslt['code'] = '111'
rslt['url'] = str(sito)
rslt['http_err'] = 'Connection refused'
else:
rslt['code'] = conn_errr
rslt['url'] = str(sito)
rslt['http_err'] = 'Connection refused'
rslt_final.append(rslt)
return rslt_final
def http_Resp(self):
rslt = {}
for sito in self.lst_urls:
try:
s = httplib2.Http()
code, resp = s.request(sito, body=None)
if code.previous:
xbmc.log("r1 http_Resp: %s %s %s %s" %
(code.status, code.reason, code.previous['status'],
code.previous['-x-permanent-redirect-url']), level=xbmc.LOGNOTICE)
rslt['code'] = code.previous['status']
rslt['redirect'] = code.previous['-x-permanent-redirect-url']
rslt['status'] = code.status
else:
rslt['code'] = code.status
except httplib2.ServerNotFoundError as msg:
# sia per mancanza di ADSL che per i siti non esistenti
rslt['code'] = -2
except socket.error as msg:
# per siti irraggiungibili senza DNS corretti
#[Errno 111] Connection refused
rslt['code'] = 111
return rslt
def view_Advise(self, txt = '' ):
"""
Avviso per utente
testConnected
"""
ip = self.check_Ip()
if ip:
txt += '\nIP: %s\n' % self.ip_addr
txt += '\nDNS: %s\n' % (self.dns)
else:
txt += '\nIP: %s' % self.ip_addr
## if self.in_addon == False and self.view_msg == True:
dialog = xbmcgui.Dialog()
dialog.textviewer(addonname, txt)
"""
def richiamato in launcher.py
"""
def test_conn(is_exit, check_dns, view_msg,
lst_urls, lst_site_check_dns, in_addon):
# debug
# import web_pdb; web_pdb.set_trace()
ktest = Kdicc(is_exit, check_dns, view_msg, lst_urls, lst_site_check_dns, in_addon)
# se non ha l'ip lo comunico all'utente
if not ktest.check_Ip():
# non permetto di entrare nell'addon
# I don't let you get into the addon
# inserire codice lingua
if view_msg == True:
ktest.view_Advise(config.get_localized_string(70720))
if ktest.is_exit == True:
exit()
# se non ha connessione ADSL lo comunico all'utente
elif not ktest.check_Adsl():
if view_msg == True:
ktest.view_Advise(config.get_localized_string(70721))
if ktest.is_exit == True:
exit()
# se ha i DNS filtrati lo comunico all'utente
elif ktest.check_dns:
if not ktest.check_Dns():
if view_msg == True:
ktest.view_Advise(config.get_localized_string(70722))
if ktest.check_Ip() and ktest.check_Adsl() and ktest.check_Dns():
return True
else:
return False
# def per la creazione del file channels.json
def check_channels(inutile=''):
"""
leggo gli host dei canali dal file channels.json
li controllo
scrivo il file channels-test.json
con i codici di errore e i nuovi url in caso di redirect
gli url DEVONO avere http(s)
Durante il controllo degli url vengono rieffettuati
i controlli di ip, asdl e dns.
Questo perchè può succedere che in un qualsiasi momento
la connessione può avere problemi.
Nel caso accada un problema, il controllo e relativa scrittura del file viene interrotto
con messaggio di avvertimento
"""
logger.info()
folderJson = xbmc.translatePath(xbmcaddon.Addon().getAddonInfo('path')).decode('utf-8')
fileJson = 'channels.json'
with open(folderJson+'/'+fileJson) as f:
data = json.load(f)
risultato = {}
for chann, host in sorted(data.items()):
ris = []
# per avere un'idea della tempistica
# utile solo se si controllano tutti i canali
# per i canali con error 522 si perdono circa 40 sec...
logger.info("check #### INIZIO #### channel - host :%s - %s " % (chann, host))
rslt = Kdicc(lst_urls = [host]).http_Resp()
# tutto ok
if rslt['code'] == 200:
risultato[chann] = host
# redirect
elif str(rslt['code']).startswith('3'):
#risultato[chann] = str(rslt['code']) +' - '+ rslt['redirect'][:-1]
if rslt['redirect'].endswith('/'):
rslt['redirect'] = rslt['redirect'][:-1]
risultato[chann] = rslt['redirect']
# sito inesistente
elif rslt['code'] == -2:
risultato[chann] = 'Host Sconosciuto - '+ str(rslt['code']) +' - '+ host
# sito non raggiungibile - probabili dns non settati
elif rslt['code'] == 111:
risultato[chann] = ['Host non raggiungibile - '+ str(rslt['code']) +' - '+ host]
else:
# altri tipi di errore
#risultato[chann] = 'Errore Sconosciuto - '+str(rslt['code']) +' - '+ host
risultato[chann] = host
logger.info("check #### FINE #### rslt :%s " % (rslt))
fileJson_test = 'channels-test.json'
# scrivo il file aggiornato
with open(folderJson+'/'+fileJson_test, 'w') as f:
data = json.dump(risultato, f, sort_keys=True, indent=4)
logger.info(data)
# codice per l'invio del file su git!!!
# 1. Bottone OKNO richiesta se vuole inviare file
# 2. Maschera per le credenziali di Github
# 3. Invio file
# 4. Esci