12f2056272
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
66 lines
2.9 KiB
Python
66 lines
2.9 KiB
Python
import importlib.util
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
spec = importlib.util.spec_from_file_location("resources", SCRIPTS / "application_recovery_resources.py")
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
class ResourcesTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.apps = {r: "synthetic-" + r for r in ("bootstrap", "reader", "custodian", "publisher")}
|
|
|
|
def grants(self, prefix, role):
|
|
policy = m.policy("synthetic-bucket", prefix, self.apps, "synthetic-owner")
|
|
return [s for s in policy["Statement"]
|
|
if s["Principal"]["SCW"] == "application_id:" + self.apps[role]]
|
|
|
|
def test_reader_only_prefixed_read_and_list(self):
|
|
rows = self.grants("kits", "reader")
|
|
actions = {a for r in rows for a in r["Action"]}
|
|
self.assertTrue({"s3:GetObject", "s3:GetObjectVersion", "s3:ListBucketVersions"} <= actions)
|
|
self.assertFalse(any("Put" in a or "Delete" in a or a == "s3:*" for a in actions))
|
|
objects = [r for r in rows if "s3:GetObject" in r["Action"]]
|
|
self.assertEqual(objects[0]["Resource"], "synthetic-bucket/kits/*")
|
|
listing = next(r for r in rows if "s3:ListBucket" in r["Action"])
|
|
self.assertEqual(listing["Condition"]["StringLike"]["s3:prefix"], ["kits/", "kits/*"])
|
|
|
|
def test_publisher_has_no_kit_grant(self):
|
|
self.assertEqual(self.grants("kits", "publisher"), [])
|
|
|
|
def test_publisher_can_abort_without_delete(self):
|
|
rows = self.grants("archives", "publisher")
|
|
actions = {a for r in rows for a in r["Action"]}
|
|
self.assertIn("s3:AbortMultipartUpload", actions)
|
|
self.assertIn("s3:PutObject", actions)
|
|
self.assertFalse(any("Delete" in a or "Policy" in a for a in actions))
|
|
|
|
def test_custodian_no_retained_delete_or_administration(self):
|
|
rows = self.grants("kits", "custodian")
|
|
actions = {a for r in rows for a in r["Action"]}
|
|
self.assertIn("s3:PutObject", actions)
|
|
self.assertFalse(any("Delete" in a or "Policy" in a or a == "s3:*" for a in actions))
|
|
|
|
def test_denial_requires_403_not_404_or_transport_error(self):
|
|
def failing(code):
|
|
raise m.NativeError(code)
|
|
self.assertTrue(m.denied(lambda: failing(403))["denied"])
|
|
for code in (400, 401, 404, 500):
|
|
with self.assertRaises(m.Refused):
|
|
m.denied(lambda code=code: failing(code))
|
|
with self.assertRaises(m.Refused):
|
|
m.denied(dict)
|
|
|
|
def test_bucket_policy_preserves_owner_and_temporary_constructor(self):
|
|
p = m.policy("b", "kits", self.apps, "o")
|
|
self.assertEqual(p["Version"], "2023-04-17")
|
|
self.assertEqual(p["Statement"][0]["Principal"], {"SCW": "user_id:o"})
|
|
self.assertTrue(all(s["Effect"] == "Allow" for s in p["Statement"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|