Implement access rule date support

Author: tw4l

pywb/warcserver/access_checker.py

@@ -173,6 +173,31 @@ def check_embargo(self, url, ts):
             actual = datetime.now(timezone.utc) - older
             return access if actual > dt else None
 
+    def check_date_access(
+        self, ts, access, default_access, rule
+    ):
+        """Return access based on date fields in access rule
+
+        If a date-based rule exists and condition is not met, return default rule
+        If no date-based rule exists, return access
+        """
+        if not rule:
+            return access
+
+        dt = timestamp_to_datetime(ts, tz_aware=True)
+
+        before_ts = rule.get('before')
+        if before_ts:
+            before = timestamp_to_datetime(before_ts, tz_aware=True)
+            return access if dt < before else default_access
+
+        after_ts = rule.get('after')
+        if after_ts:
+            after = timestamp_to_datetime(after_ts, tz_aware=True)
+            return access if dt > after else default_access
+
+        return access
+
     def create_access_aggregator(self, source_files):
         """Creates a new AccessRulesAggregator using the supplied list
         of access control file names
@@ -300,10 +325,7 @@ def wrap_iter(self, cdx_iter, acl_user):
         :param str acl_user: The user associated with this request (optional)
         :return: The wrapped cdx object iterator
         """
-        last_rule = None
-        last_url = None
-        last_user = None
-        rule = None
+        default_access = self.default_rule['access']
 
         for cdx in cdx_iter:
             url = cdx.get('url')
@@ -314,16 +336,17 @@ def wrap_iter(self, cdx_iter, acl_user):
                 yield cdx
                 continue
 
+            rule = None
             access = None
+
             if self.aggregator:
-                # TODO: optimization until date range support is included
-                if url == last_url and acl_user == last_user:
-                    rule = last_rule
-                else:
-                    rule = self.find_access_rule(url, timestamp,
-                                                 cdx.get('urlkey'),
-                                                 cdx.get('source-coll'),
-                                                 acl_user)
+                rule = self.find_access_rule(
+                    url,
+                    timestamp,
+                    cdx.get('urlkey'),
+                    cdx.get('source-coll'),
+                    acl_user
+                )
 
                 access = rule.get('access', 'exclude')
 
@@ -332,18 +355,20 @@ def wrap_iter(self, cdx_iter, acl_user):
                 if embargo_access and embargo_access != 'allow':
                     access = embargo_access
 
+            # Allow more specific rules to override embargoes
+            if access != 'exclude':
+                access = self.check_date_access(
+                    timestamp, access, default_access, rule
+                )
+
             if access == 'exclude':
                 continue
 
             if not access:
-                access = self.default_rule['access']
+                access = default_access
 
             if access == 'allow_ignore_embargo':
                 access = 'allow'
 
             cdx['access'] = access
             yield cdx
-
-            last_rule = rule
-            last_url = url
-            last_user = acl_user

sample_archive/access/after.aclj

@@ -0,0 +1 @@
+org,iana)/ - {"access": "allow", "url": "http://www.iana.org/", "after": "20140126"}

sample_archive/access/before.aclj

@@ -0,0 +1 @@
+org,iana)/ - {"access": "allow", "url": "http://www.iana.org/", "before": "20140126"}

tests/config_test_access.yaml

@@ -62,6 +62,20 @@ collections:
         acl_paths:
             - ./sample_archive/access/pywb.aclj
 
+    pywb-acl-before:
+        index_paths: ./sample_archive/cdx/
+        archive_paths: ./sample_archive/warcs/
+        default_access: block
+        acl_paths:
+            - ./sample_archive/access/before.aclj
+
+    pywb-acl-after:
+        index_paths: ./sample_archive/cdx/
+        archive_paths: ./sample_archive/warcs/
+        default_access: block
+        acl_paths:
+            - ./sample_archive/access/after.aclj
+
     pywb-wildcard-surt:
         index_paths: ./sample_archive/cdx/
         archive_paths: ./sample_archive/warcs/

tests/test_acl.py

@@ -102,3 +102,15 @@ def test_allow_all_acl_user_specific(self):
         assert 'Access Blocked' in resp.text
 
         resp = self.testapp.get('/pywb-wildcard-surt/mp_/http://example.com/', headers={"X-Pywb-Acl-User": "staff"}, status=200)
+
+    def test_acl_before(self):
+        resp = self.testapp.get('/pywb-acl-before/20140127171238mp_/http://www.iana.org/', status=451)
+        assert 'Access Blocked' in resp.text
+
+        resp = self.testapp.get('/pywb-acl-before/20140126200624mp_/http://www.iana.org/', status=200)
+        
+    def test_acl_after(self):
+        resp = self.testapp.get('/pywb-acl-after/20140126200624mp_/http://www.iana.org/', status=451)
+        assert 'Access Blocked' in resp.text
+
+        resp = self.testapp.get('/pywb-acl-after/20140127171238mp_/http://www.iana.org/', status=200)