IdentityPython / pyFF / 513

Committed 24 Jan 2019 - 20:49 coverage: 82.502%. First build

Build # 513

Build Type

Pull #158

travis-ci

Committed by

web-flow

Commit Message

utf-8 border in store implementation

Pull Request Pull Request #158: Py3 compat

Run Details

135 of 157 new or added lines in 11 files covered. (85.99%)

2744 of 3326 relevant lines covered (82.5%)

2.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.18

/src/pyff/parse.py

import os
from .utils import parse_xml, root, first_text, unicode_stream
from .constants import NS
from .logs import get_log
from xmlsec.crypto import CertDict
from datetime import datetime
from six import StringIO

__author__ = 'leifj'

log = get_log(__name__)


class ParserException(Exception):
    def __init__(self, msg, wrapped=None, data=None):
        self._wraped = wrapped
        self._data = data
        super(self.__class__, self).__init__(msg)

    def raise_wraped(self):
        raise self._wraped


class NoParser(object):
    def __init__(self):
        pass

    def magic(self, content):
        return True

    def parse(self, resource, content):
        raise ParserException("No matching parser found for %s" % resource.url)


class DirectoryParser(object):
    def __init__(self, ext):
        self.ext = ext

    def magic(self, content):
        return os.path.isdir(content)

    def _find_matching_files(self, d):
        log.debug("find files in {}".format(repr(d)))
        for top, dirs, files in os.walk(d):
            for dn in dirs:
                if dn.startswith("."):
                    dirs.remove(dn)

            for nm in files:
                if nm.endswith(self.ext):
                    fn = os.path.join(top, nm)
                    yield fn

    def parse(self, resource, content):
        resource.children = []
        n = 0
        for fn in self._find_matching_files(content):
            resource.add_child("file://" + fn)
            n += 1

        if n == 0:
            raise IOError("no entities found in {}".format(content))

        return dict()


class XRDParser(object):
    def __init__(self):
        pass

    def magic(self, content):
        return 'XRD' in content

    def parse(self, resource, content):
        info = dict()
        info['Description'] = "XRD links from {}".format(resource.url)
        t = parse_xml(unicode_stream(content))

        relt = root(t)
        for xrd in t.iter("{%s}XRD" % NS['xrd']):
            for link in xrd.findall(".//{%s}Link[@rel='%s']" % (NS['xrd'], NS['md'])):
                link_href = link.get("href")
                certs = CertDict(link)
                fingerprints = list(certs.keys())
                fp = None
                if len(fingerprints) > 0:
                    fp = fingerprints[0]
                log.debug("XRD: {} verified by {}".format(link_href, fp))
                resource.add_child(link_href, verify=fp)
        resource.last_seen = datetime.now
        resource.expire_time = None
        return info


class MDServiceListParser(object):
    def __init__(self):
        pass

    def magic(self, content):
        return 'MetadataServiceList' in content

    def parse(self, resource, content):
        info = dict()
        info['Description'] = "eIDAS MetadataServiceList from {}".format(resource.url)
        t = parse_xml(StringIO(content.encode('utf8')))
        t.xinclude()
        relt = root(t)
        info['Version'] = relt.get('Version', '0')
        info['IssueDate'] = relt.get('IssueDate')
        info['IssuerName'] = first_text(relt, "{%s}IssuerName" % NS['ser'])
        info['SchemeIdentifier'] = first_text(relt, "{%s}SchemeIdentifier" % NS['ser'])
        info['SchemeTerritory'] = first_text(relt, "{%s}SchemeTerritory" % NS['ser'])
        for mdl in relt.iter("{%s}MetadataList" % NS['ser']):
            for ml in mdl.iter("{%s}MetadataLocation" % NS['ser']):
                location = ml.get('Location')
                if location:
                    certs = CertDict(ml)
                    fingerprints = list(certs.keys())
                    fp = None
                    if len(fingerprints) > 0:
                        fp = fingerprints[0]

                    ep = ml.find("{%s}Endpoint" % NS['ser'])
                    if ep is not None and fp is not None:
                        log.debug(
                            "MetadataServiceList[{}]: {} verified by {} for country {}".format(info['SchemeTerritory'],
                                                                                               location, fp,
                                                                                               mdl.get('Territory')))
                        resource.add_child(location,
                                           verify=fp,
                                           eidas_territory=mdl.get('Territory'),
                                           eidas_endpoint_type=ep.get('EndpointType'))

        log.debug("Done parsing eIDAS MetadataServiceList")
        resource.last_seen = datetime.now
        resource.expire_time = None
        return info


_parsers = [XRDParser(), MDServiceListParser(), DirectoryParser('.xml'), NoParser()]


def add_parser(parser):
    _parsers.insert(0, parser)


def parse_resource(resource, content):
    for parser in _parsers:
        if parser.magic(content):
            return parser.parse(resource, content)

1	import os	3×
2	from .utils import parse_xml, root, first_text, unicode_stream	3×
3	from .constants import NS	3×
4	from .logs import get_log	3×
5	from xmlsec.crypto import CertDict	3×
6	from datetime import datetime	3×
7	from six import StringIO	3×
8
9	__author__ = 'leifj'	3×
10
11	log = get_log(__name__)	3×
12
13
14	class ParserException(Exception):	3×
15	def __init__(self, msg, wrapped=None, data=None):	3×
16	self._wraped = wrapped	3×
17	self._data = data	3×
18	super(self.__class__, self).__init__(msg)	3×
19
20	def raise_wraped(self):	3×
21	raise self._wraped	!
22
23
24	class NoParser(object):	3×
25	def __init__(self):	3×
26	pass	3×
27
28	def magic(self, content):	3×
29	return True	3×
30
31	def parse(self, resource, content):	3×
32	raise ParserException("No matching parser found for %s" % resource.url)	3×
33
34
35	class DirectoryParser(object):	3×
36	def __init__(self, ext):	3×
37	self.ext = ext	3×
38
39	def magic(self, content):	3×
40	return os.path.isdir(content)	3×
41
42	def _find_matching_files(self, d):	3×
43	log.debug("find files in {}".format(repr(d)))	3×
44	for top, dirs, files in os.walk(d):	3×
45	for dn in dirs:	3×
46	if dn.startswith("."):	3×
47	dirs.remove(dn)	!
48
49	for nm in files:	3×
50	if nm.endswith(self.ext):	3×
51	fn = os.path.join(top, nm)	3×
52	yield fn	3×
53
54	def parse(self, resource, content):	3×
55	resource.children = []	3×
56	n = 0	3×
57	for fn in self._find_matching_files(content):	3×
58	resource.add_child("file://" + fn)	3×
59	n += 1	3×
60
61	if n == 0:	3×
62	raise IOError("no entities found in {}".format(content))	3×
63
64	return dict()	3×
65
66
67	class XRDParser(object):	3×
68	def __init__(self):	3×
69	pass	3×
70
71	def magic(self, content):	3×
72	return 'XRD' in content	3×
73
74	def parse(self, resource, content):	3×
75	info = dict()	!
76	info['Description'] = "XRD links from {}".format(resource.url)	!
NEW 77	t = parse_xml(unicode_stream(content))	!
78
79	relt = root(t)	!
80	for xrd in t.iter("{%s}XRD" % NS['xrd']):	!
81	for link in xrd.findall(".//{%s}Link[@rel='%s']" % (NS['xrd'], NS['md'])):	!
82	link_href = link.get("href")	!
83	certs = CertDict(link)	!
84	fingerprints = list(certs.keys())	!
85	fp = None	!
86	if len(fingerprints) > 0:	!
87	fp = fingerprints[0]	!
88	log.debug("XRD: {} verified by {}".format(link_href, fp))	!
89	resource.add_child(link_href, verify=fp)	!
90	resource.last_seen = datetime.now	!
91	resource.expire_time = None	!
92	return info	!
93
94
95	class MDServiceListParser(object):	3×
96	def __init__(self):	3×
97	pass	3×
98
99	def magic(self, content):	3×
100	return 'MetadataServiceList' in content	3×
101
102	def parse(self, resource, content):	3×
103	info = dict()	!
104	info['Description'] = "eIDAS MetadataServiceList from {}".format(resource.url)	!
NEW 105	t = parse_xml(StringIO(content.encode('utf8')))	!
106	t.xinclude()	!
107	relt = root(t)	!
108	info['Version'] = relt.get('Version', '0')	!
109	info['IssueDate'] = relt.get('IssueDate')	!
110	info['IssuerName'] = first_text(relt, "{%s}IssuerName" % NS['ser'])	!
111	info['SchemeIdentifier'] = first_text(relt, "{%s}SchemeIdentifier" % NS['ser'])	!
112	info['SchemeTerritory'] = first_text(relt, "{%s}SchemeTerritory" % NS['ser'])	!
113	for mdl in relt.iter("{%s}MetadataList" % NS['ser']):	!
114	for ml in mdl.iter("{%s}MetadataLocation" % NS['ser']):	!
115	location = ml.get('Location')	!
116	if location:	!
117	certs = CertDict(ml)	!
118	fingerprints = list(certs.keys())	!
119	fp = None	!
120	if len(fingerprints) > 0:	!
121	fp = fingerprints[0]	!
122
123	ep = ml.find("{%s}Endpoint" % NS['ser'])	!
124	if ep is not None and fp is not None:	!
125	log.debug(	!
126	"MetadataServiceList[{}]: {} verified by {} for country {}".format(info['SchemeTerritory'],
127	location, fp,
128	mdl.get('Territory')))
129	resource.add_child(location,	!
130	verify=fp,
131	eidas_territory=mdl.get('Territory'),
132	eidas_endpoint_type=ep.get('EndpointType'))
133
134	log.debug("Done parsing eIDAS MetadataServiceList")	!
135	resource.last_seen = datetime.now	!
136	resource.expire_time = None	!
137	return info	!
138
139
140	_parsers = [XRDParser(), MDServiceListParser(), DirectoryParser('.xml'), NoParser()]	3×
141
142
143	def add_parser(parser):	3×
144	_parsers.insert(0, parser)	3×
145
146
147	def parse_resource(resource, content):	3×
148	for parser in _parsers:	3×
149	if parser.magic(content):	3×
150	return parser.parse(resource, content)	3×

IdentityPython / pyFF / 513

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous