ccc0f8b578
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
86 lines
4 KiB
Python
86 lines
4 KiB
Python
import copy
|
|
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"]))
|
|
|
|
def test_archive_policy_respects_native_ten_statement_limit(self):
|
|
p = m.policy("b", "archives", self.apps, "o")
|
|
self.assertLessEqual(len(p["Statement"]), 10)
|
|
shared = next(s for s in p["Statement"] if s["Sid"] == "ArchiveMetadata")
|
|
self.assertEqual(set(shared["Principal"]["SCW"]),
|
|
{"application_id:" + self.apps[r] for r in ("reader", "custodian", "publisher")})
|
|
self.assertTrue(all(a.startswith("s3:Get") and "Object" not in a for a in shared["Action"]))
|
|
|
|
def test_provider_policy_reordering_allowed_but_extra_permission_refused(self):
|
|
original = m.policy("b", "archives", self.apps, "o")
|
|
returned = copy.deepcopy(original)
|
|
returned["Statement"].reverse()
|
|
for row in returned["Statement"]:
|
|
if isinstance(row["Action"], list):
|
|
row["Action"].reverse()
|
|
self.assertEqual(m.canonical_policy(original), m.canonical_policy(returned))
|
|
returned["Statement"][0]["Action"].append("s3:DeleteObject")
|
|
self.assertNotEqual(m.canonical_policy(original), m.canonical_policy(returned))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|