You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
249 lines
9.2 KiB
249 lines
9.2 KiB
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright 2012-2016 Mir Calculate. http://www.calculate-linux.org
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
from __future__ import print_function
|
|
import urllib.request as u2
|
|
import os
|
|
import sys
|
|
import calculate.contrib
|
|
from suds.transport.http import HttpTransport
|
|
from httplib import HTTPConnection, HTTPSConnection
|
|
import socket
|
|
from calculate.core.datavars import DataVarsCore
|
|
from calculate.lib.cl_lang import setLocalTranslate
|
|
|
|
_ = lambda x: x
|
|
setLocalTranslate('cl_core3', sys.modules[__name__])
|
|
|
|
|
|
class clientHTTPSConnection(HTTPSConnection):
|
|
def __init__(self, host, port=None, key_file=None, cert_file=None,
|
|
strict=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
|
|
source_address=None, cert_path=None):
|
|
HTTPConnection.__init__(self, host, port, strict, timeout,
|
|
source_address)
|
|
self.key_file = key_file
|
|
self.cert_file = cert_file
|
|
self.cert_path = cert_path
|
|
|
|
# get filename store cert server
|
|
def cert_list(self, host, ca_certs, server_cert):
|
|
if not os.path.exists(self.ca_path):
|
|
try:
|
|
os.makedirs(self.ca_path)
|
|
except OSError:
|
|
pass
|
|
if not os.path.exists(ca_certs):
|
|
fc = open(ca_certs, "w")
|
|
fc.close()
|
|
filename = None
|
|
try:
|
|
with open(ca_certs) as fd:
|
|
t = fd.read()
|
|
# for each line
|
|
for line in t.splitlines():
|
|
# Split string into a words list
|
|
words = line.split()
|
|
if len(words) > 1:
|
|
# if first word...
|
|
if words[0] == host:
|
|
filename = words[1]
|
|
if not filename:
|
|
return None
|
|
except (IOError, IndexError):
|
|
print(_("Certificate not found on the client's side"))
|
|
return None
|
|
try:
|
|
fd = open(self.ca_path + filename, 'r')
|
|
store_cert = fd.read()
|
|
fd.close()
|
|
if store_cert == server_cert:
|
|
return filename
|
|
except IOError as e:
|
|
print(_("Failed to open the file") + "%s%s %s" % (self.ca_path,
|
|
filename, str(e)))
|
|
return None
|
|
|
|
# add certificate server in trusted
|
|
def add_server_cert(self, cert):
|
|
print(_("Untrusted server certificate!"))
|
|
import OpenSSL
|
|
|
|
certobj = OpenSSL.crypto.load_certificate(OpenSSL.SSL.FILETYPE_PEM,
|
|
cert)
|
|
print('\n' + _("Fingerprint = %s") % certobj.digest('SHA1'))
|
|
print(_("Serial number = "), certobj.get_serial_number())
|
|
Issuer = certobj.get_issuer().get_components()
|
|
print('\n' + _("Issuer"))
|
|
for i in Issuer:
|
|
print("%s : %s" % (i[0], i[1]))
|
|
Subject = certobj.get_subject().get_components()
|
|
print('\n' + _("Subject"))
|
|
for item in Subject:
|
|
print("%s : %s" % (item[0], item[1]))
|
|
|
|
choice = raw_input(
|
|
_("add this certificate to trusted and continue? y/[n]: "))
|
|
if choice in ['y', 'yes', 'Y', 'YES']:
|
|
ca_certs = self.ca_path + "cert.list"
|
|
|
|
if not os.path.exists(ca_certs):
|
|
fc = open(ca_certs, "w")
|
|
fc.close()
|
|
|
|
filename = self.host
|
|
fc = open(self.ca_path + filename, "w")
|
|
fc.write(cert)
|
|
fc.close()
|
|
with open(ca_certs) as fd:
|
|
t = fd.read()
|
|
# for each line
|
|
for line in t.splitlines():
|
|
# Split string into a words list
|
|
words = line.split()
|
|
if len(words) > 1:
|
|
# if first word...
|
|
if words[0] == self.host:
|
|
return 0
|
|
# Open file with compliance server certificates and server hostname
|
|
fcl = open(ca_certs, "a")
|
|
fcl.write(self.host + ' ' + filename + '\n')
|
|
fcl.close()
|
|
|
|
else:
|
|
sys.exit()
|
|
|
|
def connect_trusted_server(self, sock):
|
|
import ssl
|
|
|
|
if hasattr(ssl, "PROTOCOL_TLSv1_2"):
|
|
ssl_version = ssl.PROTOCOL_TLSv1_2
|
|
else:
|
|
print(_("SSL library is not support TLSv1_2"))
|
|
return 1
|
|
self.ca_path = self.cert_path + "ca/"
|
|
ca_certs = self.ca_path + "cert.list"
|
|
server_cert = ssl.get_server_certificate(addr=(self.host, self.port))
|
|
if (not hasattr(HTTPSClientsCertTransport, 'filename') or
|
|
HTTPSClientsCertTransport.filename is None):
|
|
HTTPSClientsCertTransport.filename = self.cert_list(
|
|
self.host, ca_certs, server_cert)
|
|
if HTTPSClientsCertTransport.filename:
|
|
try:
|
|
self.sock = ssl.wrap_socket(sock,
|
|
certfile=self.cert_file,
|
|
keyfile=self.key_file,
|
|
ssl_version=ssl_version,
|
|
cert_reqs=ssl.CERT_NONE)
|
|
|
|
dercert_after_connect = self.sock.getpeercert(True)
|
|
cert_after_connect = ssl.DER_cert_to_PEM_cert(
|
|
dercert_after_connect)
|
|
filename2 = self.cert_list(self.host, ca_certs,
|
|
cert_after_connect)
|
|
|
|
if not HTTPSClientsCertTransport.filename == filename2:
|
|
print('\n' + _("WARNING!!! %s trying to replace the "
|
|
"certificate!") % self.host + '\n')
|
|
self.sock.close()
|
|
return 2
|
|
return 0
|
|
except Exception:
|
|
print(_("Error. The server certificate and the private "
|
|
"key are probably invalid!"))
|
|
HTTPSClientsCertTransport.filename = None
|
|
return 1
|
|
else:
|
|
self.sock = ssl.wrap_socket(sock)
|
|
self.add_server_cert(server_cert)
|
|
|
|
def connect(self):
|
|
"""Connect to a host on a given (SSL) port."""
|
|
timeout = 15
|
|
sock = socket.create_connection((self.host, self.port),
|
|
timeout, self.source_address)
|
|
if self._tunnel_host:
|
|
self.sock = sock
|
|
self._tunnel()
|
|
|
|
clVars = DataVarsCore()
|
|
clVars.importCore()
|
|
if not clVars.flIniFile():
|
|
sys.exit(1)
|
|
import ssl
|
|
|
|
if hasattr(ssl, "PROTOCOL_TLSv1_2"):
|
|
ssl_version = ssl.PROTOCOL_TLSv1_2
|
|
else:
|
|
print(_("SSL library is not support TLSv1_2"))
|
|
sys.exit(1)
|
|
|
|
self.sock = ssl.wrap_socket(sock,
|
|
certfile=self.cert_file,
|
|
keyfile=self.key_file,
|
|
ssl_version=ssl_version,
|
|
cert_reqs=ssl.CERT_NONE)
|
|
|
|
|
|
class HTTPSClientAuthHandler(u2.HTTPSHandler):
|
|
def __init__(self, key, cert, cert_path):
|
|
u2.HTTPSHandler.__init__(self)
|
|
self.key = key
|
|
self.cert = cert
|
|
self.cert_path = cert_path
|
|
|
|
def https_open(self, req):
|
|
# Rather than pass in a reference to a connection class, we pass in
|
|
# a reference to a function which, for all intents and purposes,
|
|
# will behave as a constructor
|
|
return self.do_open(self.getConnection, req)
|
|
|
|
def getConnection(self, host, timeout=300):
|
|
return clientHTTPSConnection(host, key_file=self.key,
|
|
cert_file=self.cert,
|
|
cert_path=self.cert_path)
|
|
|
|
|
|
class HTTPSClientsCertTransport(HttpTransport):
|
|
def __init__(self, key, cert, path_to_cert, *args, **kwargs):
|
|
HttpTransport.__init__(self, *args, **kwargs)
|
|
self.key = key
|
|
self.cert = cert
|
|
self.cert_path = path_to_cert
|
|
|
|
def u2open(self, u2request):
|
|
"""
|
|
Open a connection.
|
|
@param u2request: A urllib2 request.
|
|
@type u2request: urllib2.Requet.
|
|
@return: The opened file-like urllib2 object.
|
|
@rtype: fp
|
|
"""
|
|
tm = self.options.timeout
|
|
url = u2.build_opener(HTTPSClientAuthHandler(self.key, self.cert,
|
|
self.cert_path))
|
|
|
|
# from urllib2 import URLError
|
|
# try:
|
|
if hasattr(self, "u2ver"):
|
|
if self.u2ver() < 2.6:
|
|
socket.setdefaulttimeout(tm)
|
|
return url.open(u2request)
|
|
else:
|
|
return url.open(u2request, timeout=tm)
|
|
else:
|
|
return url.open(u2request, timeout=tm)
|