add ResourceOpts, a pydantic basemodel replacing kwargs on Resource

Author: fredrikt

src/pyff/builtins.py

@@ -21,12 +21,12 @@
 from six.moves.urllib_parse import quote_plus, urlparse
 
 from pyff.pipes import registry
-
 from .constants import NS
 from .decorators import deprecated
 from .exceptions import MetadataException
 from .logs import get_log
 from .pipes import PipeException, PipelineCallback, Plumbing, pipe
+from .resource import ResourceOpts
 from .samlmd import (
     annotate_entity,
     discojson_t,
@@ -352,7 +352,7 @@ def _pipe(req, *opts):
 
 
 @pipe
-def when(req, condition, *values):
+def when(req: Plumbing.Request, condition: str, *values):
     """
     Conditionally execute part of the pipeline.
 
@@ -377,6 +377,8 @@ def when(req, condition, *values):
     followed. If 'bar' is present in the state with the value 'bill' then the other branch is followed.
     """
     c = req.state.get(condition, None)
+    if c is None:
+        log.debug(f'Condition {repr(condition)} not present in state {req.state}')
     if c is not None and (not values or _any(values, c)):
         return Plumbing(pipeline=req.args, pid="%s.when" % req.plumbing.id).iprocess(req)
     return req.t
@@ -628,32 +630,37 @@ def load(req, *opts):
         )
 
         url = r.pop(0)
-        params = {"via": [], "cleanup": [], "verify": None, "as": url}
+
+        # Copy parent node opts as a starting point
+        child_opts = req.md.rm.opts.copy(update={"via": [], "cleanup": [], "verify": None, "as": url})
 
         while len(r) > 0:
             elt = r.pop(0)
             if elt in ("as", "verify", "via", "cleanup"):
+                # These elements have an argument
                 if len(r) > 0:
-                    if elt in ("via", "cleanup"):
-                        params[elt].append(r.pop(0))
+                    value = r.pop(0)
+                    if elt == "as":
+                        child_opts.alias = value
+                    elif elt == "verify":
+                        child_opts.verify = value
+                    elif elt == "via":
+                        child_opts.via.append(PipelineCallback(value, req, store=req.md.store))
+                    elif elt == "cleanup":
+                        child_opts.cleanup.append(PipelineCallback(value, req, store=req.md.store))
                     else:
-                        params[elt] = r.pop(0)
+                        raise ValueError(f'Unhandled resource option {elt}')
                 else:
                     raise PipeException(
                         "Usage: load resource [as url] [[verify] verification] [via pipeline]* [cleanup pipeline]*"
                     )
             else:
-                params['verify'] = elt
-
-        if params['via'] is not None:
-            params['via'] = [PipelineCallback(p, req, store=req.md.store) for p in params['via']]
-
-        if params['cleanup'] is not None:
-            params['cleanup'] = [PipelineCallback(p, req, store=req.md.store) for p in params['cleanup']]
+                child_opts.verify = elt
 
-        params.update(opts)
+        # override anything in child_opts with what is in opts
+        child_opts = child_opts.copy(update=opts)
 
-        req.md.rm.add_child(url, **params)
+        req.md.rm.add_child(url, child_opts)
 
     log.debug("Refreshing all resources")
     req.md.rm.reload(fail_on_error=bool(opts['fail_on_error']))

src/pyff/parse.py

@@ -7,7 +7,7 @@
 
 from .constants import NS
 from .logs import get_log
-from .resource import Resource
+from .resource import Resource, ResourceOpts
 from .utils import find_matching_files, parse_xml, root, unicode_stream, utc_now
 
 __author__ = 'leifj'
@@ -73,7 +73,8 @@ def parse(self, resource: Resource, content: str):
         info['Expiration Time'] = 'never expires'
         n = 0
         for fn in find_matching_files(content, self.extensions):
-            resource.add_child("file://" + fn)
+            child_opts = resource.opts.copy(update={'alias': None})
+            resource.add_child("file://" + fn, child_opts)
             n += 1
 
         if n == 0:
@@ -112,7 +113,8 @@ def parse(self, resource: Resource, content: str) -> Mapping[str, Any]:
                 if len(fingerprints) > 0:
                     fp = fingerprints[0]
                 log.debug("XRD: {} verified by {}".format(link_href, fp))
-                resource.add_child(link_href, verify=fp)
+                child_opts = resource.opts.copy(update={'alias': None})
+                resource.add_child(link_href, child_opts)
         resource.last_seen = utc_now().replace(microsecond=0)
         resource.expire_time = None
         resource.never_expires = True

src/pyff/pipes.py

@@ -4,10 +4,11 @@
 """
 from __future__ import annotations
 
+import functools
 import os
 import traceback
-import functools
-from typing import Any, Dict, Optional, Callable, Type, Tuple
+from typing import Any, Callable, Dict, List, Optional
+from typing import Type
 
 import yaml
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -140,14 +141,15 @@ class PipelineCallback(object):
 
     def __init__(self, entry_point, req, store=None):
         self.entry_point = entry_point
-        self.plumbing = Plumbing(
-            req.scope_of(entry_point).plumbing.pipeline, "%s-via-%s" % (req.plumbing.id, entry_point)
-        )
+        self.plumbing = Plumbing(req.scope_of(entry_point).plumbing.pipeline, f"{req.plumbing.id}-via-{entry_point}")
         self.req = req
         self.store = store
 
     def __str__(self) -> str:
-        return f"<PipelineCallback to {self.req.plumbing}]>"
+        return f"<PipelineCallback to {self.req.plumbing}>"
+
+    def __repr__(self) -> str:
+        return str(self)
 
     def __copy__(self):
         return self
@@ -200,19 +202,19 @@ class Plumbing(object):
     would then be signed (using signer.key) and finally published in /var/metadata/public/metadata.xml
     """
 
-    def __init__(self, pipeline, pid):
+    def __init__(self, pipeline: List[Dict[str, Any]], pid: str):
         self._id = pid
         self.pipeline = pipeline
 
     def to_json(self):
         return self.pipeline
 
     @property
-    def id(self):
+    def id(self) -> str:
         return self._id
 
     @property
-    def pid(self):
+    def pid(self) -> str:
         return self._id
 
     def __iter__(self):
@@ -304,7 +306,9 @@ def iprocess(self, req: Plumbing.Request):
             try:
                 pipefn, opts, name, args = load_pipe(p)
                 log.debug(
-                    "{!s}: calling '{}' using args: {} and opts: {}".format(self.pipeline, name, repr(args), repr(opts))
+                    "{!s}: calling '{}' using args:\n{} and opts:\n{}".format(
+                        self.pipeline, name, repr(args), repr(opts)
+                    )
                 )
                 if is_text(args):
                     args = [args]
@@ -326,7 +330,16 @@ def iprocess(self, req: Plumbing.Request):
                 break
         return req.t
 
-    def process(self, md, args=None, state=None, t=None, store=None, raise_exceptions=True, scheduler=None):
+    def process(
+        self,
+        md: MDRepository,
+        args=None,
+        state: Optional[Dict[str, Any]] = None,
+        t=None,
+        store=None,
+        raise_exceptions: bool = True,
+        scheduler=None,
+    ):
         """
         The main entrypoint for processing a request pipeline. Calls the inner processor.
 

src/pyff/repo.py

@@ -2,7 +2,7 @@
 
 from .constants import config
 from .logs import get_log
-from .resource import IconHandler, Resource
+from .resource import IconHandler, Resource, ResourceOpts
 from .samlmd import entitiesdescriptor, root
 from .store import make_icon_store_instance, make_store_instance
 from .utils import is_text, make_default_scheduler
@@ -15,7 +15,7 @@ class MDRepository:
 
     def __init__(self, scheduler=None):
         random.seed(self)
-        self.rm = Resource()  # root
+        self.rm = Resource(url=None, opts=ResourceOpts())  # root
         if scheduler is None:
             scheduler = make_default_scheduler()
             scheduler.start()

src/pyff/resource.py

@@ -8,21 +8,25 @@
 import os
 import traceback
 from collections import deque
-from copy import deepcopy
 from datetime import datetime
 from threading import Condition, Lock
-from typing import Any, Callable, Deque, Dict, Optional, Tuple
+from typing import Any, Callable, Deque, Dict, List, Optional, TYPE_CHECKING, Tuple
 from urllib.parse import quote as urlescape
 
 import requests
 from requests.adapters import Response
+from pydantic import BaseModel, Field
 
 from .constants import config
 from .exceptions import ResourceException
 from .fetch import make_fetcher
 from .logs import get_log
 from .utils import Watchable, hex_digest, img_to_data, non_blocking_lock, resource_string, safe_write, url_get, utc_now
 
+if TYPE_CHECKING:
+    from .parse import PyffParser
+    from .pipes import PipelineCallback
+
 requests.packages.urllib3.disable_warnings()
 
 log = get_log(__name__)
@@ -131,11 +135,32 @@ def i_handle(self, t: Resource, url=None, response=None, exception=None, last_fe
             t.info['Exception'] = ex
 
 
+class ResourceOpts(BaseModel):
+    alias: Optional[str] = Field(None, alias='as')  # TODO: Rename to 'name'?
+    # a certificate (file) or a SHA1 fingerprint to use for signature verification
+    verify: Optional[str] = None
+    via: List[Any] = Field([])  # list of PipelineCallback
+    # A list of PipelineCallback that can be used to pre-process parsed metadata before validation. Use as a clue-bat.
+    cleanup: List[Any] = Field([])  # list of PipelineCallback
+    fail_on_error: bool = False
+    # remove invalid EntityDescriptor elements rather than raise an error
+    filter_invalid: bool = True
+    # set to False to turn off all XML schema validation
+    validate_schema: bool = Field(True, alias='validate')
+
+    def to_dict(self) -> Dict[str, Any]:
+        res = self.dict()
+        # Compensate for the 'alias' field options
+        res['as'] = res.pop('alias')
+        res['validate'] = res.pop('validate_schema')
+        return res
+
+
 class Resource(Watchable):
-    def __init__(self, url=None, **kwargs):
+    def __init__(self, url: Optional[str], opts: ResourceOpts):
         super().__init__()
         self.url: str = url
-        self.opts = kwargs
+        self.opts = opts
         self.t = None
         self.type = "text/plain"
         self.etag = None
@@ -148,12 +173,6 @@ def __init__(self, url=None, **kwargs):
         self._setup()
 
     def _setup(self):
-        self.opts.setdefault('cleanup', [])
-        self.opts.setdefault('via', [])
-        self.opts.setdefault('fail_on_error', False)
-        self.opts.setdefault('verify', None)
-        self.opts.setdefault('filter_invalid', True)
-        self.opts.setdefault('validate', True)
         if self.url is not None:
             if "://" not in self.url:
                 pth = os.path.abspath(self.url)
@@ -178,20 +197,20 @@ def local_copy_fn(self):
         return os.path.join(config.local_copy_dir, urlescape(self.url))
 
     @property
-    def post(self):
-        return self.opts['via']
+    def post(self) -> List['PipelineCallback']:
+        return self.opts.via
 
-    def add_via(self, callback):
-        self.opts['via'].append(callback)
+    def add_via(self, callback: 'PipelineCallback') -> None:
+        self.opts.via.append(callback)
 
     @property
-    def cleanup(self):
-        return self.opts['cleanup']
+    def cleanup(self) -> List['PipelineCallback']:
+        return self.opts.cleanup
 
     def __str__(self):
         return "Resource {} expires at {} using ".format(
             self.url if self.url is not None else "(root)", self.expire_time
-        ) + ",".join(["{}={}".format(k, v) for k, v in list(self.opts.items())])
+        ) + ",".join(["{}={}".format(k, v) for k, v in sorted(list(self.opts.dict().items()))])
 
     def reload(self, fail_on_error=False):
         with non_blocking_lock(self.lock):
@@ -252,25 +271,24 @@ def _replace(self, r: Resource) -> None:
                 return
         raise ValueError("Resource {} not present - use add_child".format(r.url))
 
-    def add_child(self, url: str, **kwargs) -> Resource:
-        opts = deepcopy(self.opts)
-        if 'as' in opts:
-            del opts['as']
-        opts.update(kwargs)
-        r = Resource(url, **opts)
+    def add_child(self, url: str, opts: ResourceOpts) -> Resource:
+        r = Resource(url, opts)
         if r in self.children:
+            log.debug(f'\n\n{self}:\nURL {url}\nReplacing child {r}')
             self._replace(r)
         else:
+            log.debug(f'\n\n{self}:\nURL {url}\nAdding child {r}')
+            if not r.opts.via:
+                log.debug('Empty Via')
             self.children.append(r)
 
         return r
 
     @property
     def name(self) -> Optional[str]:
-        if 'as' in self.opts:
-            return self.opts['as']
-        else:
-            return self.url
+        if self.opts.alias:
+            return self.opts.alias
+        return self.url
 
     @property
     def info(self):
@@ -359,7 +377,7 @@ def load_resource(self, getter: Callable[[str], Response]) -> Tuple[Optional[str
 
         return data, status, info
 
-    def parse(self, getter):
+    def parse(self, getter: Callable[[str], Response]) -> Deque[Resource]:
         data, status, info = self.load_resource(getter)
         info['State'] = 'Parsing'
         # local import to avoid circular import
@@ -375,10 +393,10 @@ def parse(self, getter):
 
         info['State'] = 'Parsed'
         if self.t is not None:
-            if self.post and isinstance(self.post, list):
+            if self.post:
                 for cb in self.post:
                     if self.t is not None:
-                        self.t = cb(self.t, **self.opts)
+                        self.t = cb(self.t, self.opts.dict())
 
             if self.is_expired():
                 info['Expired'] = True