Implement visibility-aware S3 ACL controls for product security
This update enhances product file security by automatically managing S3 object ACLs based on product visibility settings: - Added get_s3_acl_for_file_key() method to determine appropriate ACL per file type and visibility - Added update_s3_acls() method to batch update all product file permissions - Added set_visibility() method that updates visibility and synchronizes S3 ACLs - Updated product edit view to use new ACL-aware visibility setting - Added comprehensive unit tests covering all visibility scenarios - Created migration script to fix existing product S3 ACLs Security model: - Private products: All files (product, preview, thumbnails) are private - Public/unlisted products: Product files remain private, public files (previews/thumbnails) are public-read - All product downloads continue to use presigned URLs for access control
This commit is contained in:
parent
afa08780bf
commit
6e0bad873a
4 changed files with 275 additions and 4 deletions
|
|
@ -484,6 +484,54 @@ class Product(RBase, Base):
|
|||
def stamp_updated_timestamp(self):
|
||||
self.updated_timestamp = now_timestamp()
|
||||
|
||||
def get_s3_acl_for_file_key(self, file_key):
|
||||
"""Return appropriate S3 ACL based on product visibility and file type."""
|
||||
# Public files (thumbnails, previews) are always public-read for public/unlisted products
|
||||
if file_key in self.file_public_keys:
|
||||
if self.is_private:
|
||||
return "private"
|
||||
else:
|
||||
return "public-read"
|
||||
|
||||
# Product files are always private regardless of visibility
|
||||
# Access is controlled via presigned URLs
|
||||
return "private"
|
||||
|
||||
def update_s3_acls(self, s3_client, bucket_name):
|
||||
"""Update S3 ACLs for all product files based on current visibility."""
|
||||
|
||||
# Update ACLs for all file types
|
||||
for file_key in self.file_keys:
|
||||
if file_key in self.extensions:
|
||||
s3_key = getattr(self, f's3_key_{file_key}', None)
|
||||
if s3_key is None and file_key == 'product':
|
||||
s3_key = self.s3_key
|
||||
elif s3_key is None and file_key == 'preview':
|
||||
s3_key = self.s3_key_preview
|
||||
elif s3_key is None and file_key.startswith('thumbnail'):
|
||||
s3_key = self.s3_key_thumbnail(file_key)
|
||||
|
||||
if s3_key:
|
||||
try:
|
||||
acl = self.get_s3_acl_for_file_key(file_key)
|
||||
s3_client.put_object_acl(
|
||||
Bucket=bucket_name,
|
||||
Key=s3_key,
|
||||
ACL=acl
|
||||
)
|
||||
except Exception as e:
|
||||
# Log error but don't fail the visibility change
|
||||
print(f"Failed to update S3 ACL for {s3_key}: {e}")
|
||||
|
||||
def set_visibility(self, new_visibility, s3_client=None, bucket_name=None):
|
||||
"""Set product visibility and update S3 ACLs accordingly."""
|
||||
old_visibility = self.visibility
|
||||
self.visibility = new_visibility
|
||||
|
||||
# Update S3 ACLs if client provided and visibility changed
|
||||
if s3_client and bucket_name and old_visibility != new_visibility:
|
||||
self.update_s3_acls(s3_client, bucket_name)
|
||||
|
||||
|
||||
def get_all_products(dbsession):
|
||||
"""
|
||||
|
|
|
|||
106
make_post_sell/scripts/fix_s3_acls.py
Normal file
106
make_post_sell/scripts/fix_s3_acls.py
Normal file
|
|
@ -0,0 +1,106 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Migration script to fix S3 object ACLs based on product visibility.
|
||||
|
||||
This script updates S3 object permissions for all existing products to match
|
||||
their current visibility settings. Run this after implementing the new
|
||||
visibility-based S3 ACL controls.
|
||||
|
||||
Usage:
|
||||
python -m make_post_sell.scripts.fix_s3_acls development.ini
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pyramid.paster import bootstrap
|
||||
from make_post_sell.models.product import get_all_products
|
||||
|
||||
|
||||
def fix_product_s3_acls(env, dry_run=True):
|
||||
"""Fix S3 ACLs for all products based on their visibility settings."""
|
||||
|
||||
request = env['request']
|
||||
dbsession = request.dbsession
|
||||
|
||||
# Get S3 client and bucket name
|
||||
s3_client = request.secure_uploads_client
|
||||
bucket_name = request.app["bucket.secure_uploads"]
|
||||
|
||||
# Get all products
|
||||
products = get_all_products(dbsession).all()
|
||||
|
||||
print(f"Found {len(products)} products to process")
|
||||
print(f"Bucket: {bucket_name}")
|
||||
print(f"Dry run: {dry_run}")
|
||||
print("-" * 50)
|
||||
|
||||
processed = 0
|
||||
errors = 0
|
||||
|
||||
for product in products:
|
||||
try:
|
||||
print(f"Processing product {product.id} ({product.title})")
|
||||
print(f" Visibility: {product.human_visibility} ({product.visibility})")
|
||||
|
||||
# Check which files exist for this product
|
||||
existing_files = []
|
||||
for file_key in product.file_keys:
|
||||
if file_key in product.extensions:
|
||||
existing_files.append(file_key)
|
||||
|
||||
if not existing_files:
|
||||
print(f" No files found, skipping")
|
||||
continue
|
||||
|
||||
print(f" Files: {existing_files}")
|
||||
|
||||
# Show what ACLs would be applied
|
||||
for file_key in existing_files:
|
||||
acl = product.get_s3_acl_for_file_key(file_key)
|
||||
print(f" {file_key}: {acl}")
|
||||
|
||||
if not dry_run:
|
||||
# Actually update the ACLs
|
||||
product.update_s3_acls(s3_client, bucket_name)
|
||||
print(f" ✓ Updated ACLs")
|
||||
else:
|
||||
print(f" (Dry run - no changes made)")
|
||||
|
||||
processed += 1
|
||||
|
||||
except Exception as e:
|
||||
print(f" ✗ Error processing product {product.id}: {e}")
|
||||
errors += 1
|
||||
|
||||
print()
|
||||
|
||||
print("-" * 50)
|
||||
print(f"Summary:")
|
||||
print(f" Processed: {processed}")
|
||||
print(f" Errors: {errors}")
|
||||
print(f" Dry run: {dry_run}")
|
||||
|
||||
|
||||
def main():
|
||||
"""Main entry point."""
|
||||
parser = argparse.ArgumentParser(description='Fix S3 ACLs for products based on visibility')
|
||||
parser.add_argument('config_uri', help='Configuration file (e.g., development.ini)')
|
||||
parser.add_argument('--execute', action='store_true',
|
||||
help='Actually make changes (default is dry run)')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Bootstrap the Pyramid application
|
||||
with bootstrap(args.config_uri) as env:
|
||||
try:
|
||||
fix_product_s3_acls(env, dry_run=not args.execute)
|
||||
except KeyboardInterrupt:
|
||||
print("\nOperation cancelled by user")
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
|
@ -157,3 +157,121 @@ class TestCoupon(unittest.TestCase):
|
|||
# test inverse.
|
||||
self.assertFalse(self.disabled_coupon.is_valid)
|
||||
self.assertFalse(self.disabled_coupon.is_expired)
|
||||
|
||||
|
||||
class TestProductS3Security(unittest.TestCase):
|
||||
"""Test S3 security controls for product visibility."""
|
||||
|
||||
def setUp(self):
|
||||
from ..models.product import Product
|
||||
from ..models.shop import Shop
|
||||
|
||||
# Create test shop and product
|
||||
self.shop = Shop(
|
||||
"test-shop",
|
||||
"555-555-5555",
|
||||
"123 Test St",
|
||||
"Test shop description",
|
||||
)
|
||||
self.product = Product("Test Product", "Test product description")
|
||||
self.product.shop = self.shop
|
||||
|
||||
# Mock file metadata to simulate uploaded files
|
||||
self.product.json_file_metadata = '{"extensions": {"product": "pdf", "preview": "jpg", "thumbnail1": "jpg"}}'
|
||||
self.product._file_metadata = {
|
||||
"extensions": {"product": "pdf", "preview": "jpg", "thumbnail1": "jpg"}
|
||||
}
|
||||
|
||||
def test_get_s3_acl_for_public_product(self):
|
||||
"""Test S3 ACL logic for public products."""
|
||||
self.product.visibility = 1 # public
|
||||
|
||||
# Product files should always be private
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("product"), "private")
|
||||
|
||||
# Public files should be public-read for public products
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("preview"), "public-read")
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("thumbnail1"), "public-read")
|
||||
|
||||
def test_get_s3_acl_for_unlisted_product(self):
|
||||
"""Test S3 ACL logic for unlisted products."""
|
||||
self.product.visibility = 2 # unlisted
|
||||
|
||||
# Product files should always be private
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("product"), "private")
|
||||
|
||||
# Public files should be public-read for unlisted products
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("preview"), "public-read")
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("thumbnail1"), "public-read")
|
||||
|
||||
def test_get_s3_acl_for_private_product(self):
|
||||
"""Test S3 ACL logic for private products."""
|
||||
self.product.visibility = 0 # private
|
||||
|
||||
# All files should be private for private products
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("product"), "private")
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("preview"), "private")
|
||||
self.assertEqual(self.product.get_s3_acl_for_file_key("thumbnail1"), "private")
|
||||
|
||||
@mock.patch('builtins.print') # Mock print to avoid test output
|
||||
def test_update_s3_acls_with_mock_client(self, mock_print):
|
||||
"""Test S3 ACL updates with mocked S3 client."""
|
||||
# Create mock S3 client
|
||||
mock_s3_client = mock.Mock()
|
||||
bucket_name = "test-bucket"
|
||||
|
||||
self.product.visibility = 1 # public
|
||||
|
||||
# Should call put_object_acl for each file
|
||||
self.product.update_s3_acls(mock_s3_client, bucket_name)
|
||||
|
||||
# Verify put_object_acl was called for each file type
|
||||
expected_calls = len(self.product.extensions)
|
||||
self.assertEqual(mock_s3_client.put_object_acl.call_count, expected_calls)
|
||||
|
||||
def test_set_visibility_without_s3_client(self):
|
||||
"""Test set_visibility without S3 client (no ACL updates)."""
|
||||
old_visibility = self.product.visibility
|
||||
new_visibility = 0 # private
|
||||
|
||||
# Should update visibility without error
|
||||
self.product.set_visibility(new_visibility)
|
||||
self.assertEqual(self.product.visibility, new_visibility)
|
||||
|
||||
@mock.patch('builtins.print') # Mock print to avoid test output
|
||||
def test_set_visibility_with_s3_client(self, mock_print):
|
||||
"""Test set_visibility with S3 client (should update ACLs)."""
|
||||
mock_s3_client = mock.Mock()
|
||||
bucket_name = "test-bucket"
|
||||
|
||||
old_visibility = 1 # public
|
||||
new_visibility = 0 # private
|
||||
self.product.visibility = old_visibility
|
||||
|
||||
# Should update visibility and call update_s3_acls
|
||||
self.product.set_visibility(new_visibility, mock_s3_client, bucket_name)
|
||||
|
||||
self.assertEqual(self.product.visibility, new_visibility)
|
||||
# Verify S3 ACL update was called
|
||||
expected_calls = len(self.product.extensions)
|
||||
self.assertEqual(mock_s3_client.put_object_acl.call_count, expected_calls)
|
||||
|
||||
def test_visibility_property_helpers(self):
|
||||
"""Test visibility helper properties work correctly."""
|
||||
# Test public
|
||||
self.product.visibility = 1
|
||||
self.assertTrue(self.product.is_public)
|
||||
self.assertFalse(self.product.is_unlisted)
|
||||
self.assertFalse(self.product.is_private)
|
||||
|
||||
# Test unlisted
|
||||
self.product.visibility = 2
|
||||
self.assertFalse(self.product.is_public)
|
||||
self.assertTrue(self.product.is_unlisted)
|
||||
self.assertFalse(self.product.is_private)
|
||||
|
||||
# Test private
|
||||
self.product.visibility = 0
|
||||
self.assertFalse(self.product.is_public)
|
||||
self.assertFalse(self.product.is_unlisted)
|
||||
self.assertTrue(self.product.is_private)
|
||||
|
|
|
|||
|
|
@ -237,7 +237,7 @@ def product_edit(request):
|
|||
|
||||
if visibility != product.visibility:
|
||||
product_modified = True
|
||||
product.visibility = visibility
|
||||
product.set_visibility(visibility, request.secure_uploads_client, request.app["bucket.secure_uploads"])
|
||||
request.session.flash(("You updated the product's visibility.", "success"))
|
||||
|
||||
if price != product.price:
|
||||
|
|
@ -266,9 +266,8 @@ def product_edit(request):
|
|||
|
||||
# This is how we protect our digital downloads while using the
|
||||
# same bucket for thumbnails and previews related to the product.
|
||||
acl = "private"
|
||||
if file_key in product.file_public_keys:
|
||||
acl = "public-read"
|
||||
# Use the new visibility-aware ACL method
|
||||
acl = product.get_s3_acl_for_file_key(file_key)
|
||||
if product.is_not_sellable:
|
||||
acl = "public-read"
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue