Support for retention policies on containers (#3501)

- [x] ability to specify a retention period on a container, which applies to newly-created blobs
- [x] specify default retention periods in templates from CLI side 

There's a small breaking change to the Python JobHelper class.

.github/workflows/ci.yml

@@ -123,7 +123,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v4
         with:
-          python-version: 3.7
+          python-version: "3.10"
       - name: lint
         shell: bash
         run: src/ci/check-check-pr.sh
@@ -137,7 +137,7 @@ jobs:
         shell: bash
       - uses: actions/setup-python@v4
         with:
-          python-version: 3.7
+          python-version: "3.10"
       - uses: actions/download-artifact@v3
         with:
           name: artifact-onefuzztypes
@@ -190,7 +190,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v4
         with:
-          python-version: 3.8
+          python-version: "3.10"
       - name: lint
         shell: bash
         run: |
@@ -208,7 +208,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v4
         with:
-          python-version: 3.8
+          python-version: "3.10"
       - name: lint
         shell: bash
         run: |
@@ -224,7 +224,7 @@ jobs:
       - run: src/ci/set-versions.sh
       - uses: actions/setup-python@v4
         with:
-          python-version: 3.7
+          python-version: "3.10"
       - run: src/ci/onefuzztypes.sh
       - uses: actions/upload-artifact@v3
         with:
@@ -481,7 +481,7 @@ jobs:
           path: artifacts
       - uses: actions/setup-python@v4
         with:
-          python-version: 3.7
+          python-version: "3.10"
       - name: Lint
         shell: bash
         run: |

src/ApiService/ApiService/FeatureFlags.cs

@@ -8,4 +8,5 @@ public static class FeatureFlagConstants {
     public const string EnableBlobRetentionPolicy = "EnableBlobRetentionPolicy";
     public const string EnableDryRunBlobRetention = "EnableDryRunBlobRetention";
     public const string EnableWorkItemCreation = "EnableWorkItemCreation";
+    public const string EnableContainerRetentionPolicies = "EnableContainerRetentionPolicies";
 }

src/ApiService/ApiService/Functions/QueueFileChanges.cs

@@ -1,5 +1,6 @@
 using System.Text.Json;
 using System.Text.Json.Nodes;
+using System.Threading.Tasks;
 using Azure.Core;
 using Microsoft.Azure.Functions.Worker;
 using Microsoft.Extensions.Logging;
@@ -54,14 +55,16 @@ public async Async.Task Run(
             return;
         }
 
+        var storageAccount = new ResourceIdentifier(topicElement.GetString()!);
+
         try {
             // Setting isLastRetryAttempt to false will rethrow any exceptions
             // With the intention that the azure functions runtime will handle requeing
             // the message for us. The difference is for the poison queue, we're handling the
             // requeuing ourselves because azure functions doesn't support retry policies
             // for queue based functions.
 
-            var result = await FileAdded(fileChangeEvent, isLastRetryAttempt: false);
+            var result = await FileAdded(storageAccount, fileChangeEvent, isLastRetryAttempt: false);
             if (!result.IsOk && result.ErrorV.Code == ErrorCode.ADO_WORKITEM_PROCESSING_DISABLED) {
                 await RequeueMessage(msg, TimeSpan.FromDays(1));
             }
@@ -71,16 +74,47 @@ public async Async.Task Run(
         }
     }
 
-    private async Async.Task<OneFuzzResultVoid> FileAdded(JsonDocument fileChangeEvent, bool isLastRetryAttempt) {
+    private async Async.Task<OneFuzzResultVoid> FileAdded(ResourceIdentifier storageAccount, JsonDocument fileChangeEvent, bool isLastRetryAttempt) {
         var data = fileChangeEvent.RootElement.GetProperty("data");
         var url = data.GetProperty("url").GetString()!;
         var parts = url.Split("/").Skip(3).ToList();
 
-        var container = parts[0];
+        var container = Container.Parse(parts[0]);
         var path = string.Join('/', parts.Skip(1));
 
-        _log.LogInformation("file added : {Container} - {Path}", container, path);
-        return await _notificationOperations.NewFiles(Container.Parse(container), path, isLastRetryAttempt);
+        _log.LogInformation("file added : {Container} - {Path}", container.String, path);
+
+        var (_, result) = await (
+            ApplyRetentionPolicy(storageAccount, container, path),
+            _notificationOperations.NewFiles(container, path, isLastRetryAttempt));
+
+        return result;
+    }
+
+    private async Async.Task<bool> ApplyRetentionPolicy(ResourceIdentifier storageAccount, Container container, string path) {
+        if (await _context.FeatureManagerSnapshot.IsEnabledAsync(FeatureFlagConstants.EnableContainerRetentionPolicies)) {
+            // default retention period can be applied to the container
+            // if one exists, we will set the expiry date on the newly-created blob, if it doesn't already have one
+            var account = await _storage.GetBlobServiceClientForAccount(storageAccount);
+            var containerClient = account.GetBlobContainerClient(container.String);
+            var containerProps = await containerClient.GetPropertiesAsync();
+            var retentionPeriod = RetentionPolicyUtils.GetContainerRetentionPeriodFromMetadata(containerProps.Value.Metadata);
+            if (!retentionPeriod.IsOk) {
+                _log.LogError("invalid retention period: {Error}", retentionPeriod.ErrorV);
+            } else if (retentionPeriod.OkV is TimeSpan period) {
+                var blobClient = containerClient.GetBlobClient(path);
+                var tags = (await blobClient.GetTagsAsync()).Value.Tags;
+                var expiryDate = DateTime.UtcNow + period;
+                var tag = RetentionPolicyUtils.CreateExpiryDateTag(DateOnly.FromDateTime(expiryDate));
+                if (tags.TryAdd(tag.Key, tag.Value)) {
+                    _ = await blobClient.SetTagsAsync(tags);
+                    _log.LogInformation("applied container retention policy ({Policy}) to {Path}", period, path);
+                    return true;
+                }
+            }
+        }
+
+        return false;
     }
 
     private async Async.Task RequeueMessage(string msg, TimeSpan? visibilityTimeout = null) {

src/ApiService/ApiService/OneFuzzTypes/Enums.cs

@@ -50,6 +50,7 @@ public enum ErrorCode {
     ADO_WORKITEM_PROCESSING_DISABLED = 494,
     ADO_VALIDATION_INVALID_PATH = 495,
     ADO_VALIDATION_INVALID_PROJECT = 496,
+    INVALID_RETENTION_PERIOD = 497,
     // NB: if you update this enum, also update enums.py
 }
 

src/ApiService/ApiService/onefuzzlib/NotificationOperations.cs

@@ -22,13 +22,12 @@ public NotificationOperations(ILogger<NotificationOperations> log, IOnefuzzConte
 
     }
     public async Async.Task<OneFuzzResultVoid> NewFiles(Container container, string filename, bool isLastRetryAttempt) {
-        var result = OneFuzzResultVoid.Ok;
-
         // We don't want to store file added events for the events container because that causes an infinite loop
         if (container == WellKnownContainers.Events) {
-            return result;
+            return Result.Ok();
         }
 
+        var result = OneFuzzResultVoid.Ok;
         var notifications = GetNotifications(container);
         var hasNotifications = await notifications.AnyAsync();
         var reportOrRegression = await _context.Reports.GetReportOrRegression(container, filename, expectReports: hasNotifications);

src/ApiService/ApiService/onefuzzlib/RetentionPolicy.cs

@@ -0,0 +1,43 @@
+using System.Xml;
+
+namespace Microsoft.OneFuzz.Service;
+
+
+public interface IRetentionPolicy {
+    DateOnly GetExpiryDate();
+}
+
+public class RetentionPolicyUtils {
+    public const string EXPIRY_TAG = "Expiry";
+    public static KeyValuePair<string, string> CreateExpiryDateTag(DateOnly expiryDate) =>
+        new(EXPIRY_TAG, expiryDate.ToString());
+
+    public static DateOnly? GetExpiryDateTagFromTags(IDictionary<string, string>? blobTags) {
+        if (blobTags != null &&
+            blobTags.TryGetValue(EXPIRY_TAG, out var expiryTag) &&
+            !string.IsNullOrWhiteSpace(expiryTag) &&
+            DateOnly.TryParse(expiryTag, out var expiryDate)) {
+            return expiryDate;
+        }
+        return null;
+    }
+
+    public static string CreateExpiredBlobTagFilter() => $@"""{EXPIRY_TAG}"" <= '{DateOnly.FromDateTime(DateTime.UtcNow)}'";
+
+    // NB: this must match the value used on the CLI side
+    public const string CONTAINER_RETENTION_KEY = "onefuzz_retentionperiod";
+
+    public static OneFuzzResult<TimeSpan?> GetContainerRetentionPeriodFromMetadata(IDictionary<string, string>? containerMetadata) {
+        if (containerMetadata is not null &&
+            containerMetadata.TryGetValue(CONTAINER_RETENTION_KEY, out var retentionString) &&
+            !string.IsNullOrWhiteSpace(retentionString)) {
+            try {
+                return Result.Ok<TimeSpan?>(XmlConvert.ToTimeSpan(retentionString));
+            } catch (Exception ex) {
+                return Error.Create(ErrorCode.INVALID_RETENTION_PERIOD, ex.Message);
+            }
+        }
+
+        return Result.Ok<TimeSpan?>(null);
+    }
+}

src/cli/examples/domato.py

@@ -67,7 +67,7 @@ def upload_to_fuzzer_container(of: Onefuzz, fuzzer_name: str, fuzzer_url: str) -
 
 
 def upload_to_setup_container(of: Onefuzz, helper: JobHelper, setup_dir: str) -> None:
-    setup_sas = of.containers.get(helper.containers[ContainerType.setup]).sas_url
+    setup_sas = of.containers.get(helper.container_name(ContainerType.setup)).sas_url
     if AZCOPY_PATH is None:
         raise Exception("missing azcopy")
     command = [AZCOPY_PATH, "sync", setup_dir, setup_sas]
@@ -143,13 +143,16 @@ def main() -> None:
     helper.create_containers()
     helper.setup_notifications(notification_config)
     upload_to_setup_container(of, helper, args.setup_dir)
-    add_setup_script(of, helper.containers[ContainerType.setup])
+    add_setup_script(of, helper.container_name(ContainerType.setup))
 
     containers = [
-        (ContainerType.setup, helper.containers[ContainerType.setup]),
-        (ContainerType.crashes, helper.containers[ContainerType.crashes]),
-        (ContainerType.reports, helper.containers[ContainerType.reports]),
-        (ContainerType.unique_reports, helper.containers[ContainerType.unique_reports]),
+        (ContainerType.setup, helper.container_name(ContainerType.setup)),
+        (ContainerType.crashes, helper.container_name(ContainerType.crashes)),
+        (ContainerType.reports, helper.container_name(ContainerType.reports)),
+        (
+            ContainerType.unique_reports,
+            helper.container_name(ContainerType.unique_reports),
+        ),
     ]
 
     of.logger.info("Creating generic_crash_report task")
@@ -164,11 +167,11 @@ def main() -> None:
 
     containers = [
         (ContainerType.tools, Container(FUZZER_NAME)),
-        (ContainerType.setup, helper.containers[ContainerType.setup]),
-        (ContainerType.crashes, helper.containers[ContainerType.crashes]),
+        (ContainerType.setup, helper.container_name(ContainerType.setup)),
+        (ContainerType.crashes, helper.container_name(ContainerType.crashes)),
         (
             ContainerType.readonly_inputs,
-            helper.containers[ContainerType.readonly_inputs],
+            helper.container_name(ContainerType.readonly_inputs),
         ),
     ]
 

src/cli/examples/honggfuzz.py

@@ -88,13 +88,16 @@ def main() -> None:
     if args.inputs:
         helper.upload_inputs(args.inputs)
 
-    add_setup_script(of, helper.containers[ContainerType.setup])
+    add_setup_script(of, helper.container_name(ContainerType.setup))
 
     containers = [
-        (ContainerType.setup, helper.containers[ContainerType.setup]),
-        (ContainerType.crashes, helper.containers[ContainerType.crashes]),
-        (ContainerType.reports, helper.containers[ContainerType.reports]),
-        (ContainerType.unique_reports, helper.containers[ContainerType.unique_reports]),
+        (ContainerType.setup, helper.container_name(ContainerType.setup)),
+        (ContainerType.crashes, helper.container_name(ContainerType.crashes)),
+        (ContainerType.reports, helper.container_name(ContainerType.reports)),
+        (
+            ContainerType.unique_reports,
+            helper.container_name(ContainerType.unique_reports),
+        ),
     ]
 
     of.logger.info("Creating generic_crash_report task")
@@ -109,11 +112,11 @@ def main() -> None:
 
     containers = [
         (ContainerType.tools, Container("honggfuzz")),
-        (ContainerType.setup, helper.containers[ContainerType.setup]),
-        (ContainerType.crashes, helper.containers[ContainerType.crashes]),
+        (ContainerType.setup, helper.container_name(ContainerType.setup)),
+        (ContainerType.crashes, helper.container_name(ContainerType.crashes)),
         (
             ContainerType.inputs,
-            helper.containers[ContainerType.inputs],
+            helper.container_name(ContainerType.inputs),
         ),
     ]
 

src/cli/examples/llvm-source-coverage/source-coverage-libfuzzer.py

@@ -74,15 +74,15 @@ def main() -> None:
     helper.create_containers()
 
     of.containers.files.upload_file(
-        helper.containers[ContainerType.tools], f"{args.tools}/source-coverage.sh"
+        helper.container_name(ContainerType.tools), f"{args.tools}/source-coverage.sh"
     )
 
     containers = [
-        (ContainerType.setup, helper.containers[ContainerType.setup]),
-        (ContainerType.analysis, helper.containers[ContainerType.analysis]),
-        (ContainerType.tools, helper.containers[ContainerType.tools]),
+        (ContainerType.setup, helper.container_name(ContainerType.setup)),
+        (ContainerType.analysis, helper.container_name(ContainerType.analysis)),
+        (ContainerType.tools, helper.container_name(ContainerType.tools)),
         # note, analysis is typically for crashes, but this is analyzing inputs
-        (ContainerType.crashes, helper.containers[ContainerType.inputs]),
+        (ContainerType.crashes, helper.container_name(ContainerType.inputs)),
     ]
 
     of.logger.info("Creating generic_analysis task")

src/cli/examples/llvm-source-coverage/source-coverage.py

@@ -61,15 +61,15 @@ def main() -> None:
         helper.upload_inputs(args.inputs)
 
     of.containers.files.upload_file(
-        helper.containers[ContainerType.tools], f"{args.tools}/source-coverage.sh"
+        helper.container_name(ContainerType.tools), f"{args.tools}/source-coverage.sh"
     )
 
     containers = [
-        (ContainerType.setup, helper.containers[ContainerType.setup]),
-        (ContainerType.analysis, helper.containers[ContainerType.analysis]),
-        (ContainerType.tools, helper.containers[ContainerType.tools]),
+        (ContainerType.setup, helper.container_name(ContainerType.setup)),
+        (ContainerType.analysis, helper.container_name(ContainerType.analysis)),
+        (ContainerType.tools, helper.container_name(ContainerType.tools)),
         # note, analysis is typically for crashes, but this is analyzing inputs
-        (ContainerType.crashes, helper.containers[ContainerType.inputs]),
+        (ContainerType.crashes, helper.container_name(ContainerType.inputs)),
     ]
 
     of.logger.info("Creating generic_analysis task")