1
0
mirror of https://github.com/webrecorder/pywb.git synced 2025-03-14 15:53:28 +01:00
pywb/tests/test_acl_manager.py
Ilya Kreymer f07d35709a
Access Control Improvements: Embargo + ACL User Support (#642)
* embargo: add support for per-collection date range embargo with embargo options of 'before', 'after', 'newer' and 'older'
'before' and 'after' accept a timestamp
'newer' and 'older' options configured with a dictionary consisting of any combo of 'years', 'months', 'days'
add basic test for each embargo option

* acl/embargo work:
- support acl access value 'allow_ignore_embargo' for overriding embargo
- support 'user' in acl setting, matched with value of 'X-Pywb-ACL-User' header
- support passing through 'X-Pywb-ACL-User' setting to warcserver
- aclmanager: support -u/--user param for adding, removing and matching rules
- tests: add test for 'allow_ignore_embargo', user-specific acl rule matching

* docs: add docs for new embargo system!

* docs: add info on how to configure ACL header with short examples to usage page.
sample-deploy: add examples of configuring X-pywb-ACL-user header based on IP for nginx and apache sample deployments

* docs: fix access control page header, text tweaks

* bump version to 2.6.0b0
2021-05-18 20:09:18 -07:00

212 lines
6.8 KiB
Python

import os
from .base_config_test import BaseConfigTest, CollsDirMixin, fmod
from pywb.manager.manager import main as wb_manager
from pytest import raises
# ============================================================================
class TestACLManager(CollsDirMixin, BaseConfigTest):
@classmethod
def setup_class(cls):
super(TestACLManager, cls).setup_class('config_test_access.yaml')
cls.acl_filename = os.path.join(cls.root_dir, 'acl', 'test.aclj')
@classmethod
def teardown_class(cls):
super(TestACLManager, cls).teardown_class()
try:
os.remove(cls.acl_filename)
except:
pass
def test_acl_add_err_wrong_access(self):
with raises(SystemExit):
wb_manager(['acl', 'add', self.acl_filename, 'http://example.com/', 'access'])
def test_acl_add(self):
wb_manager(['acl', 'add', self.acl_filename, 'http://example.com/', 'allow'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_acl_add_surt(self):
wb_manager(['acl', 'add', self.acl_filename, 'com,example,', 'exclude'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example, - {"access": "exclude", "url": "com,example,"}
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_acl_add_with_user(self):
wb_manager(['acl', 'add', self.acl_filename, 'http://example.com/', 'block', '-u', 'public'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example, - {"access": "exclude", "url": "com,example,"}
com,example)/ - {"access": "block", "url": "http://example.com/", "user": "public"}
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_acl_list(self, capsys):
wb_manager(['acl', 'list', self.acl_filename])
out, err = capsys.readouterr()
assert out == """\
Rules for %s from %s:
com,example, - {"access": "exclude", "url": "com,example,"}
com,example)/ - {"access": "block", "url": "http://example.com/", "user": "public"}
com,example)/ - {"access": "allow", "url": "http://example.com/"}
""" % (self.acl_filename, self.acl_filename)
def test_acl_list_err_no_such_file(self):
with raises(SystemExit):
wb_manager(['acl', 'list', self.acl_filename + '2'])
def test_acl_match(self, capsys):
wb_manager(['acl', 'match', self.acl_filename, 'http://abc.example.com/foo'])
out, err = capsys.readouterr()
assert out == """\
Matched rule:
com,example, - {"access": "exclude", "url": "com,example,"}
"""
def test_acl_match_user(self, capsys):
wb_manager(['acl', 'match', self.acl_filename, 'http://example.com/foo', '-u', 'public'])
out, err = capsys.readouterr()
assert out == """\
Matched rule:
com,example)/ - {"access": "block", "url": "http://example.com/", "user": "public"}
"""
def test_acl_match_unknown_user(self, capsys):
wb_manager(['acl', 'match', self.acl_filename, 'http://example.com/foo', '-u', 'data'])
out, err = capsys.readouterr()
assert out == """\
Matched rule:
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_acl_match_default_user(self, capsys):
wb_manager(['acl', 'match', self.acl_filename, 'http://example.com/foo'])
out, err = capsys.readouterr()
assert out == """\
Matched rule:
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_remove_acl(self):
wb_manager(['acl', 'remove', self.acl_filename, 'com,example,'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example)/ - {"access": "block", "url": "http://example.com/", "user": "public"}
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_remove_acl_user(self):
wb_manager(['acl', 'remove', self.acl_filename, 'com,example)/', '-u', 'public'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_acl_add_exact(self):
wb_manager(['acl', 'add', '--exact-match', self.acl_filename, 'example.com', 'block'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example)/### - {"access": "block", "url": "example.com"}
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_remove_acl_exact(self):
wb_manager(['acl', 'remove', '-e', self.acl_filename, 'https://example.com/'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_validate_and_sort_acl(self):
with open(self.acl_filename, 'at') as fh:
fh.write('com,example)/subpath - {"access": "block", "url": "http://example.com/subpath"}\n')
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example)/ - {"access": "allow", "url": "http://example.com/"}
com,example)/subpath - {"access": "block", "url": "http://example.com/subpath"}
"""
wb_manager(['acl', 'validate', self.acl_filename])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
com,example)/subpath - {"access": "block", "url": "http://example.com/subpath"}
com,example)/ - {"access": "allow", "url": "http://example.com/"}
"""
def test_importtxt_acl(self, capsys):
name = os.path.join(self.root_dir, 'excludes.txt')
with open(name, 'wt') as exc:
exc.write('http://iana.org/\n')
exc.write('http://example.com/subpath/another\n')
exc.write('http://example.co/foo/\n')
exc.write('http://example.com/\n')
wb_manager(['acl', 'importtxt', self.acl_filename, name, 'exclude'])
with open(self.acl_filename, 'rt') as fh:
assert fh.read() == """\
org,iana)/ - {"access": "exclude", "url": "http://iana.org/"}
com,example)/subpath/another - {"access": "exclude", "url": "http://example.com/subpath/another"}
com,example)/subpath - {"access": "block", "url": "http://example.com/subpath"}
com,example)/ - {"access": "exclude", "url": "http://example.com/"}
co,example)/foo - {"access": "exclude", "url": "http://example.co/foo/"}
"""
out, err = capsys.readouterr()
assert 'Added or replaced 4 rules from {0}'.format(name) in out, out
os.remove(name)
def test_import_errors(self):
# missing access mode
with raises(SystemExit):
wb_manager(['acl', 'importtxt', self.acl_filename, 'foo'])
# no such file
with raises(SystemExit):
wb_manager(['acl', 'importtxt', self.acl_filename, 'foo', 'exclude'])