[elbe-devel] [PATCH 16/20] debinstaller elbexml: refactor proxy handling and urlopener
Torben Hohn
torben.hohn at linutronix.de
Fri Oct 12 11:28:02 CEST 2018
proxy dictionary setup and authentication is needed in
multiple places.
Isolate ElbeOpener() and get_proxy_dict() from the code in
ElbeXML.validate_apt_sources() and move it into elbepack.debinstaller
get_proxy_dict() can also generate dicts that can fed to
the env_add environment. Special care is taken because
of defaults, because None means use env var default, and
{} means empty proxy settings.
also make "elbe init" use the new code.
Signed-off-by: Torben Hohn <torben.hohn at linutronix.de>
---
elbepack/commands/init.py | 24 +++++--------
elbepack/debinstaller.py | 87 +++++++++++++++++++++++++++++++++++++++++------
elbepack/elbexml.py | 34 ++----------------
3 files changed, 89 insertions(+), 56 deletions(-)
diff --git a/elbepack/commands/init.py b/elbepack/commands/init.py
index e43307e4..27c2332d 100644
--- a/elbepack/commands/init.py
+++ b/elbepack/commands/init.py
@@ -15,7 +15,7 @@ from optparse import OptionParser
from elbepack.treeutils import etree
from elbepack.validate import validate_xml
-from elbepack.debinstaller import copy_kinitrd, NoKinitrdException
+from elbepack.debinstaller import copy_kinitrd, NoKinitrdException, get_proxy_dict
from elbepack.xmldefaults import ElbeDefaults
from elbepack.version import elbe_version
from elbepack.templates import write_template, get_initvm_preseed
@@ -112,14 +112,13 @@ def run_command(argv):
defs = ElbeDefaults(buildtype)
- http_proxy = ""
- if os.getenv("http_proxy"):
- http_proxy = os.getenv("http_proxy")
- elif opt.proxy:
- http_proxy = opt.proxy
- elif xml.has("initvm/mirror/primary_proxy"):
- http_proxy = xml.text("initvm/mirror/primary_proxy")
- http_proxy = http_proxy.strip().replace("LOCALMACHINE", "localhost")
+ proxy_dict = get_proxy_dict(xml.node("/initvm"),
+ proxy_override=opt.proxy,
+ initvm=False)
+ if proxy_dict:
+ http_proxy = proxy_dict['http']
+ else:
+ http_proxy = os.getenv("http_proxy", default="")
if opt.cdrom:
mirror = xml.node("initvm/mirror")
@@ -159,13 +158,8 @@ def run_command(argv):
"preseed": get_initvm_preseed(xml),
"cfg": cfg}
- if http_proxy != "":
- os.putenv("http_proxy", http_proxy)
- os.putenv("https_proxy", http_proxy)
- os.putenv("no_proxy", "localhost,127.0.0.1")
-
try:
- copy_kinitrd(xml.node("/initvm"), out_path)
+ copy_kinitrd(xml.node("/initvm"), out_path, proxy_dict)
except NoKinitrdException as e:
print("Failure to download kernel/initrd debian Package:")
print("")
diff --git a/elbepack/debinstaller.py b/elbepack/debinstaller.py
index de4f3696..2960bbb7 100644
--- a/elbepack/debinstaller.py
+++ b/elbepack/debinstaller.py
@@ -9,7 +9,11 @@ import sys
import os
import re
-from urllib2 import urlopen
+from urllib2 import (ProxyHandler,
+ build_opener,
+ HTTPPasswordMgrWithDefaultRealm,
+ HTTPBasicAuthHandler)
+from urlparse import urlparse
from shutil import copyfileobj, copyfile
from gpgme import Context
@@ -98,16 +102,78 @@ def setup_apt_keyring(gpg_home, keyring_fname):
except CommandError:
print('adding keyring "%s" to keyring "%s" failed' % (key, ring_path))
-def download(url, local_fname):
+def get_proxy_dict(prj,
+ proxy_default_to_env=True,
+ proxy_override=None,
+ initvm=True,
+ for_env=False):
+ if initvm:
+ no_value = "10.0.2.2,localhost,127.0.0.1"
+ else:
+ no_value = "localhost,127.0.0.1"
+
+ if for_env:
+ key_prefix = "_proxy"
+ else:
+ key_prefix = ""
+
+ if proxy_override:
+ proxy = proxy_override
+ elif prj.has("mirror/primary_proxy"):
+ proxy = prj.text(
+ "mirror/primary_proxy").strip().replace("LOCALMACHINE",
+ "10.0.2.2")
+ else:
+ if for_env:
+ if proxy_default_to_env:
+ return {}
+ else:
+ proxy = ""
+ no_value = ""
+ else:
+ if proxy_default_to_env:
+ return None
+ else:
+ return {}
+
+ return {"no"+key_prefix: no_value,
+ "http"+key_prefix: proxy,
+ "https"+key_prefix: proxy}
+
+def download(opener, url, local_fname):
+ rf = opener.open(url, None, 10)
try:
- rf = urlopen(url, None, 10)
with open(local_fname, "w") as wf:
copyfileobj(rf, wf)
finally:
rf.close()
+class ElbeOpener(object):
+ def __init__(self, url, proxy_dict):
+ proxy_handler = ProxyHandler(proxy_dict)
+ handlers = [proxy_handler]
+
+ parsed = urlparse(url)
+ if parsed.password:
+ passman = HTTPPasswordMgrWithDefaultRealm()
+ authhandler = HTTPBasicAuthHandler(passman)
+ passman.add_password(None, url, parsed.username, parsed.password)
+ handlers.append(authhandler)
+
+ self.opener = build_opener(*handlers)
+
+ def open(self, fullurl, data=None, timeout=None):
+ return self.opener.open(fullurl, data, timeout)
+
+ def download(self, url, local_fname):
+ rf = self.open(url, None, 10)
+ try:
+ with open(local_fname, "w") as wf:
+ copyfileobj(rf, wf)
+ finally:
+ rf.close()
-def download_release(tmp, base_url):
+def download_release(tmp, base_url, opener):
# setup gpg context, for verifying
# the Release.gpg signature.
@@ -118,14 +184,14 @@ def download_release(tmp, base_url):
# download the Relase file to a tmp file,
# because we need it 2 times
- download(base_url + "Release", tmp.fname('Release'))
+ opener.download(base_url + "Release", tmp.fname('Release'))
# validate signature.
# open downloaded plaintext file, and
# use the urlopen object of the Release.gpg
# directtly.
try:
- sig = urlopen(base_url + 'Release.gpg', None, 10)
+ sig = opener.open(base_url + 'Release.gpg', None, 10)
with tmp.open("Release", "r") as signed:
overall_status = OverallStatus()
@@ -142,7 +208,7 @@ def download_release(tmp, base_url):
finally:
sig.close()
-def download_kinitrd(tmp, suite, mirror):
+def download_kinitrd(tmp, suite, mirror, opener):
base_url = "%s/dists/%s/" % (mirror, suite)
installer_path = "main/installer-amd64/current/images/"
@@ -150,7 +216,7 @@ def download_kinitrd(tmp, suite, mirror):
# download release file and check
# signature
- download_release(tmp, base_url)
+ download_release(tmp, base_url, opener)
# parse Release file, and remember hashvalues
# we are interested in
@@ -196,7 +262,7 @@ def get_primary_mirror(prj):
return mirror.replace("LOCALMACHINE", "10.0.2.2")
-def copy_kinitrd(prj, target_dir):
+def copy_kinitrd(prj, target_dir, proxy_dict):
suite = prj.text("suite")
@@ -211,7 +277,8 @@ def copy_kinitrd(prj, target_dir):
os.path.join(target_dir, "initrd.gz"))
else:
mirror = get_primary_mirror(prj)
- download_kinitrd(tmp, suite, mirror)
+ opener = ElbeOpener(mirror, proxy_dict)
+ download_kinitrd(tmp, suite, mirror, opener)
copyfile(tmp.fname("initrd.gz"),
os.path.join(target_dir, "initrd.gz"))
diff --git a/elbepack/elbexml.py b/elbepack/elbexml.py
index 42cd8bc7..8d401077 100644
--- a/elbepack/elbexml.py
+++ b/elbepack/elbexml.py
@@ -18,6 +18,7 @@ from tempfile import NamedTemporaryFile
from elbepack.treeutils import etree
from elbepack.validate import validate_xml
from elbepack.xmldefaults import ElbeDefaults
+from elbepack.debinstaller import ElbeOpener, get_proxy_dict
from elbepack.version import elbe_version, is_devel
@@ -197,9 +198,6 @@ class ElbeXML(object):
def validate_apt_sources(self, url_validation, buildtype):
- # pylint: disable=too-many-locals
- # pylint: disable=too-many-branches
-
slist = self.create_apt_sources_list()
sources_lines = slist.split('\n')
@@ -236,35 +234,9 @@ class ElbeXML(object):
if not self.prj:
return
- if self.prj.has("mirror/primary_proxy"):
- proxy = self.prj.text(
- "mirror/primary_proxy").strip().replace("LOCALMACHINE",
- "10.0.2.2")
- proxies = {"no": "10.0.2.2,localhost,127.0.0.1",
- "http": proxy,
- "https": proxy}
- else:
- proxies = {}
-
- proxy_handler = urllib2.ProxyHandler(proxies)
-
- passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
- authhandler = urllib2.HTTPBasicAuthHandler(passman)
-
- opener = urllib2.build_opener(proxy_handler, authhandler)
-
for r in repos:
- if '@' in r["url"]:
- t = r["url"].split('@')
- if '://' in t[0]:
- scheme, auth = t[0].split('://')
- scheme = scheme + '://'
- else:
- scheme = ''
- auth = t[0]
- r["url"] = scheme + t[1]
- usr, passwd = auth.split(':')
- passman.add_password(None, r["url"], usr, passwd)
+ proxy_dict = get_proxy_dict(self.prj, proxy_default_to_env=False)
+ opener = ElbeOpener(r["url"], proxy_dict)
if not self.validate_repo(opener, r, url_validation):
raise ValidationError(
["Repository %s can not be validated" % r["url"]])
--
2.11.0
More information about the elbe-devel
mailing list