Merge pull request #3643 from OpenNeuroOrg/private-exports

Export all datasets and manage object tags for access control

Author: nellh

services/datalad/datalad_service/common/openneuro.py

@@ -73,3 +73,30 @@ def update_file_check(dataset_path, commit, references, bad_files, remote=None):
     except requests.exceptions.HTTPError as e:
         logging.error(e)
         logging.error(req.text)
+
+
+def is_public_dataset(dataset_id):
+    """Ask the OpenNeuro API if this dataset is public."""
+    try:
+        logger = logging.getLogger(__name__)
+        response = requests.post(
+            GRAPHQL_ENDPOINT,
+            json={
+                'query': 'query($datasetId: ID!) { dataset(id: $datasetId) { public } }',
+                'variables': {'datasetId': dataset_id},
+            },
+            headers={'authorization': f'Bearer {generate_service_token(dataset_id)}'},
+        )
+        response.raise_for_status()
+        data = response.json()
+        if 'errors' in data:
+            logger.error(
+                f'GraphQL error checking public status for {dataset_id}: {data["errors"]}'
+            )
+            raise Exception(
+                f'GraphQL error checking public status for {dataset_id}: {data["errors"]}'
+            )
+        return data.get('data', {}).get('dataset', {}).get('public', False)
+    except requests.exceptions.RequestException as e:
+        logger.error(f'Failed to check public status for {dataset_id}: {e}')
+        raise

services/datalad/datalad_service/common/s3.py

@@ -72,6 +72,7 @@ def backup_remote_env():
 def setup_s3_sibling(dataset_path):
     """Add a sibling for an S3 bucket publish."""
     # Public remote
+    # TODO set ['x-amz-tagging=access=private'] if tagging is supported in git-annex
     subprocess.run(
         ['git-annex', 'initremote', get_s3_remote()]
         + generate_s3_annex_options(dataset_path),

services/datalad/datalad_service/handlers/dataset.py

@@ -41,7 +41,7 @@ async def on_post(self, req, resp, dataset):
                 author = pygit2.Signature(name, email)
             else:
                 author = None
-            hexsha = await create_dataset(self.store, dataset, author)
+            hexsha = await create_dataset(ds_path, author)
             resp.media = {'hexsha': hexsha}
             resp.status = falcon.HTTP_OK
 

services/datalad/datalad_service/handlers/publish.py

@@ -1,6 +1,12 @@
 import falcon
 
-from datalad_service.tasks.publish import create_remotes_and_export
+from taskiq_pipelines import Pipeline
+
+from datalad_service.broker import broker
+from datalad_service.tasks.publish import (
+    create_remotes_and_export,
+    set_s3_access_tag,
+)
 
 
 class PublishResource:
@@ -11,6 +17,11 @@ def __init__(self, store):
 
     async def on_post(self, req, resp, dataset):
         dataset_path = self.store.get_dataset_path(dataset)
-        await create_remotes_and_export.kiq(dataset_path)
+        # Pipeline create and export -> set access tag to public
+        await (
+            Pipeline(broker, create_remotes_and_export)
+            .call_after(set_s3_access_tag, dataset=dataset, value='public')
+            .kiq(dataset_path)  # create_remotes_and_export
+        )
         resp.media = {}
         resp.status = falcon.HTTP_OK

services/datalad/datalad_service/tasks/dataset.py

@@ -13,6 +13,7 @@
 
 from datalad_service.common.annex import init_annex
 from datalad_service.common.git import git_commit, COMMITTER_EMAIL, COMMITTER_NAME
+from datalad_service.tasks.publish import create_remotes
 
 # A list of patterns to avoid annexing in BIDS datasets
 GIT_ATTRIBUTES = """* annex.backend=SHA256E
@@ -44,12 +45,11 @@ def create_datalad_config(dataset_path):
         configfile.write(config)
 
 
-async def create_dataset(store, dataset, author=None, initial_head='main'):
+async def create_dataset(dataset_path, author=None, initial_head='main'):
     """Create a DataLad git-annex repo for a new dataset.
 
     initial_head is only meant for tests and is overridden by the implementation of git_commit
     """
-    dataset_path = store.get_dataset_path(dataset)
     if os.path.isdir(dataset_path):
         raise Exception('Dataset already exists')
     if not author:
@@ -70,6 +70,7 @@ async def create_dataset(store, dataset, author=None, initial_head='main'):
         '[OpenNeuro] Dataset created',
         parents=[],
     )
+    create_remotes(dataset_path)
     return str(repo.head.target)
 
 

services/datalad/datalad_service/tasks/publish.py

@@ -21,7 +21,7 @@
 from datalad_service.config import GCP_ACCESS_KEY_ID
 from datalad_service.config import GCP_SECRET_ACCESS_KEY
 from datalad_service.common.annex import get_tag_info, is_git_annex_remote
-from datalad_service.common.openneuro import clear_dataset_cache
+from datalad_service.common.openneuro import clear_dataset_cache, is_public_dataset
 from datalad_service.common.git import git_show, git_tag, git_tag_tree
 from datalad_service.common.github import github_export
 from datalad_service.common.s3 import (
@@ -126,6 +126,8 @@ async def export_dataset(
         if tags:
             new_tag = tags[-1].name
             await s3_export(dataset_path, get_s3_remote(), new_tag)
+            if not is_public_dataset(dataset_id):
+                await set_s3_access_tag(dataset_id, 'private')
             await s3_backup_push(dataset_path)
             # Once all S3 tags are exported, update GitHub
             if github_enabled:
@@ -274,3 +276,48 @@ async def annex_drop(fsck_success, dataset_path, branch):
         await run_check(
             ['git-annex', 'drop', '--branch', branch], dataset_path, env=env
         )
+
+
+async def set_remote_public(dataset):
+    """Clear x-amz-meta-access when a dataset is made public."""
+    # If git-annex supports tags in the future, we'd modify this here.
+    # await run_check(
+    #    ['git-annex', 'enableremote', get_s3_remote(), 'x-amz-tagging=access=public'],
+    #    dataset_path,
+    # )
+    await set_s3_access_tag(dataset, 'public')
+
+
+@broker.task
+async def set_s3_access_tag(dataset, value='private'):
+    """Set access tag on all versions of all files."""
+    client = boto3.client(
+        's3',
+        aws_access_key_id=AWS_ACCESS_KEY_ID,
+        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
+    )
+    s3_bucket = get_s3_bucket()
+    paginator = client.get_paginator('list_object_versions')
+    for page in paginator.paginate(Bucket=s3_bucket, Prefix=f'{dataset}/'):
+        for version in page.get('Versions', []):
+            key = version['Key']
+            version_id = version['VersionId']
+            try:
+                response = client.get_object_tagging(
+                    Bucket=s3_bucket, Key=key, VersionId=version_id
+                )
+                tag_set = response.get('TagSet', [])
+            except client.exceptions.ClientError as e:
+                if e.response['Error']['Code'] == 'NoSuchTagSet':
+                    tag_set = []
+                else:
+                    raise
+            # Remove any existing access tag and add the new one
+            new_tags = [tag for tag in tag_set if tag['Key'] != 'access']
+            new_tags.append({'Key': 'access', 'Value': value})
+            client.put_object_tagging(
+                Bucket=s3_bucket,
+                Key=key,
+                VersionId=version_id,
+                Tagging={'TagSet': new_tags},
+            )

services/datalad/tests/conftest.py

@@ -4,6 +4,7 @@
 import json
 import subprocess
 import random
+from unittest import mock
 
 import pytest
 from falcon import testing
@@ -249,3 +250,11 @@ async def async_noop_validator(*args, **kwargs):
     monkeypatch.setattr(
         'datalad_service.tasks.files.validate_dataset', async_noop_validator
     )
+
+
+@pytest.fixture(autouse=True)
+def is_public_dataset_mock(monkeypatch):
+    def _mock(dataset_id):
+        return True
+
+    monkeypatch.setattr('datalad_service.tasks.publish.is_public_dataset', _mock)

services/datalad/tests/test_common_s3.py

@@ -37,6 +37,26 @@ def test_s3_annex_options(monkeypatch):
     # Check prefix and bucket strings are interpolated right
     assert 'fileprefix=test00001/' in options
     assert 'bucket=a-fake-test-public-bucket' in options
+    # Check that the private tag is applied by default
+    # assert 'x-amz-tagging=access=private' in options
+
+
+def test_s3_annex_backup_options(monkeypatch):
+    monkeypatch.setattr(
+        datalad_service.config, 'AWS_S3_PUBLIC_BUCKET', 'a-fake-test-public-bucket'
+    )
+    monkeypatch.setattr(datalad_service.config, 'GCP_S3_BACKUP_BUCKET', 'backup-bucket')
+    options = generate_s3_annex_options(
+        '/tmp/dataset/does/not/exist/test00001', backup=True
+    )
+    assert 'type=S3' in options
+    # Verify public=no (ACL deprecation)
+    assert 'public=no' in options
+    # Verify autoenable=true is not present
+    assert 'autoenable=true' not in options
+    # Check prefix and bucket strings are interpolated right
+    assert 'fileprefix=test00001/' in options
+    assert 'bucket=backup-bucket' in options
 
 
 def test_update_s3_sibling(monkeypatch, no_init_remote, new_dataset):

services/datalad/tests/test_datalad.py

@@ -16,7 +16,8 @@
 async def test_create_dataset(datalad_store):
     ds_id = 'ds000002'
     author = pygit2.Signature('test author', '[email protected]')
-    await create_dataset(datalad_store, ds_id, author)
+    ds_path = os.path.join(datalad_store.annex_path, ds_id)
+    await create_dataset(ds_path, author)
     ds = Dataset(os.path.join(datalad_store.annex_path, ds_id))
     assert ds.repo is not None
     # Verify the dataset is created with datalad config
@@ -52,10 +53,11 @@ async def test_create_dataset_master(datalad_store):
 async def test_create_dataset_unusual_default_branch(datalad_store):
     ds_id = 'ds000026'
     author = pygit2.Signature('test author', '[email protected]')
+    ds_path = os.path.join(datalad_store.annex_path, ds_id)
     # Create dataset will commit data and this should fail since HEAD is something 'unusual'
     # (such as the git-annex branch as a plausible example)
     with pytest.raises(OpenNeuroGitError) as e:
-        await create_dataset(datalad_store, ds_id, author, 'unusual')
+        await create_dataset(ds_path, author, 'unusual')
 
 
 async def test_delete_dataset(datalad_store, new_dataset):