GRR port to Spanner

.github/workflows/build.yml

@@ -3,7 +3,12 @@ on: [push, pull_request]
 env:
   GCS_BUCKET: autobuilds.grr-response.com
   GCS_BUCKET_OPENAPI: autobuilds-grr-openapi
-  DOCKER_REPOSITORY: ghcr.io/google/grr
+  DOCKER_REPOSITORY: ghcr.io/${{ github.repository_owner }}/grr
+  #SPANNER_EMULATOR_HOST: localhost:9010
+  #SPANNER_DATABASE: grr-database
+  #SPANNER_INSTANCE: grr-instance
+  #SPANNER_PROJECT_ID: spanner-emulator-project
+  #PROJECT_ID: spanner-emulator-project
 jobs:
   test-devenv:
     runs-on: ubuntu-22.04
@@ -44,6 +49,27 @@ jobs:
           fi
           python3 -m venv --system-site-packages "${HOME}/INSTALL"
           travis/install.sh
+      #- name: 'Install Cloud SDK'
+      #  uses: google-github-actions/setup-gcloud@v2
+      #  with:
+      #    install_components: 'beta,pubsub-emulator,cloud-spanner-emulator'
+      #- name: 'Start Spanner emulator'
+      #  run: |
+      #    gcloud config configurations create emulator
+      #    gcloud config set auth/disable_credentials true
+      #    gcloud config set project ${{ env.SPANNER_PROJECT_ID }}
+      #    gcloud config set api_endpoint_overrides/spanner http://localhost:9020/
+      #    gcloud emulators spanner start &
+      #    gcloud spanner instances create ${{ env.SPANNER_INSTANCE }} \
+      #      --config=emulator-config --description="Spanner Test Instance" --nodes=1
+      #    sudo apt-get update
+      #    sudo apt install -y protobuf-compiler libprotobuf-dev
+      #    # Verify that we can create the Spanner Instance and Database
+      #    cd grr/server/grr_response_server/databases
+      #    ./spanner_setup.sh
+      #    cd -
+      #    # Remove the Database for a clean test env
+      #    gcloud spanner databases delete ${{ env.SPANNER_DATABASE }} --instance=${{ env.SPANNER_INSTANCE }} --quiet
       - name: Test
         run: |
           source "${HOME}/INSTALL/bin/activate"

.gitignore

@@ -36,3 +36,6 @@ grr/server/grr_response_server/gui/ui/.angular/
 docker_config_files/*.pem
 compose.watch.yaml
 Dockerfile.client
+
+grr/server/grr_response_server/databases/spanner_grr.pb
+spanner-setup.txt

compose.testing.yaml

@@ -1,14 +1,14 @@
 
 services:
   grr-admin-ui:
-    image: ghcr.io/google/grr:testing
+    image: ghcr.io/google/grr:test
 
   grr-fleetspeak-frontend:
-    image: ghcr.io/google/grr:testing
+    image: ghcr.io/google/grr:test
 
   grr-worker:
-    image: ghcr.io/google/grr:testing
+    image: ghcr.io/google/grr:test
 
   grr-client:
-    image: ghcr.io/google/grr:testing
+    image: ghcr.io/google/grr:test
     privileged: true

grr/core/grr_response_core/config/data_store.py

@@ -121,3 +121,33 @@
         "Only used when Blobstore.implementation is GCSBlobStore."
     ),
 )
+
+# Spanner configuration.
+config_lib.DEFINE_string(
+    "Spanner.project",
+    default=None,
+    help=(
+        "GCP Project where the Spanner Instance is located."
+    ),
+)
+config_lib.DEFINE_string(
+    "Spanner.instance",
+    default="grr-instance",
+    help="The Spanner Instance for GRR.")
+
+config_lib.DEFINE_string(
+    "Spanner.database",
+    default="grr-database",
+    help="The Spanner Database for GRR.")
+
+config_lib.DEFINE_integer(
+    "Spanner.flow_processing_threads_min",
+    default=1,
+    help="The minimum number of flow-processing worker threads.",
+)
+
+config_lib.DEFINE_integer(
+    "Spanner.flow_processing_threads_max",
+    default=20,
+    help="The maximum number of flow-processing worker threads.",
+)

grr/server/grr_response_server/databases/db_cronjob_test.py

@@ -191,7 +191,7 @@ def testCronJobLeasing(self):
     job = self._CreateCronJob()
     self.db.WriteCronJob(job)
 
-    current_time = rdfvalue.RDFDatetime.FromSecondsSinceEpoch(10000)
+    current_time = rdfvalue.RDFDatetime.Now()
     lease_time = rdfvalue.Duration.From(5, rdfvalue.MINUTES)
     with test_lib.FakeTime(current_time):
       leased = self.db.LeaseCronJobs(lease_time=lease_time)
@@ -232,7 +232,7 @@ def testCronJobLeasingByID(self):
     for j in jobs:
       self.db.WriteCronJob(j)
 
-    current_time = rdfvalue.RDFDatetime.FromSecondsSinceEpoch(10000)
+    current_time = rdfvalue.RDFDatetime.Now()
     lease_time = rdfvalue.Duration.From(5, rdfvalue.MINUTES)
     with test_lib.FakeTime(current_time):
       leased = self.db.LeaseCronJobs(
@@ -254,7 +254,7 @@ def testCronJobReturning(self):
     with self.assertRaises(ValueError):
       self.db.ReturnLeasedCronJobs([job])
 
-    current_time = rdfvalue.RDFDatetime.FromSecondsSinceEpoch(10000)
+    current_time = rdfvalue.RDFDatetime.Now()
     with test_lib.FakeTime(current_time):
       leased = self.db.LeaseCronJobs(
           cronjob_ids=[leased_job.cron_job_id],
@@ -276,14 +276,14 @@ def testCronJobReturningMultiple(self):
     for j in jobs:
       self.db.WriteCronJob(j)
 
-    current_time = rdfvalue.RDFDatetime.FromSecondsSinceEpoch(10000)
+    current_time = rdfvalue.RDFDatetime.Now()
     with test_lib.FakeTime(current_time):
       leased = self.db.LeaseCronJobs(
           lease_time=rdfvalue.Duration.From(5, rdfvalue.MINUTES)
       )
       self.assertLen(leased, 3)
 
-    current_time = rdfvalue.RDFDatetime.FromSecondsSinceEpoch(10001)
+    current_time = rdfvalue.RDFDatetime.Now()
     with test_lib.FakeTime(current_time):
       unleased_jobs = self.db.LeaseCronJobs(
           lease_time=rdfvalue.Duration.From(5, rdfvalue.MINUTES)
@@ -292,7 +292,7 @@ def testCronJobReturningMultiple(self):
 
       self.db.ReturnLeasedCronJobs(leased)
 
-    current_time = rdfvalue.RDFDatetime.FromSecondsSinceEpoch(10002)
+    current_time = rdfvalue.RDFDatetime.Now()
     with test_lib.FakeTime(current_time):
       leased = self.db.LeaseCronJobs(
           lease_time=rdfvalue.Duration.From(5, rdfvalue.MINUTES)

grr/server/grr_response_server/databases/db_flows_test.py

@@ -2270,15 +2270,16 @@ def testWritesAndReadsSingleFlowResultOfSingleType(self):
     sample_result = flows_pb2.FlowResult(client_id=client_id, flow_id=flow_id)
     sample_result.payload.Pack(jobs_pb2.ClientSummary(client_id=client_id))
 
-    with test_lib.FakeTime(42):
+    current_time = rdfvalue.RDFDatetime.Now()
+    with test_lib.FakeTime(current_time):
       self.db.WriteFlowResults([sample_result])
 
     results = self.db.ReadFlowResults(client_id, flow_id, 0, 100)
     self.assertLen(results, 1)
     self.assertEqual(results[0].payload, sample_result.payload)
     self.assertEqual(
         rdfvalue.RDFDatetime.FromMicrosecondsSinceEpoch(results[0].timestamp),
-        rdfvalue.RDFDatetime.FromSecondsSinceEpoch(42),
+        current_time,
     )
 
   def testWritesAndReadsRDFStringFlowResult(self):
@@ -2335,7 +2336,8 @@ def testReadResultsRestoresAllFlowResultsFields(self):
     )
     sample_result.payload.Pack(jobs_pb2.ClientSummary(client_id=client_id))
 
-    with test_lib.FakeTime(42):
+    current_time = rdfvalue.RDFDatetime.Now()
+    with test_lib.FakeTime(current_time):
       self.db.WriteFlowResults([sample_result])
 
     results = self.db.ReadFlowResults(client_id, flow_id, 0, 100)
@@ -2346,7 +2348,7 @@ def testReadResultsRestoresAllFlowResultsFields(self):
     self.assertEqual(results[0].payload, sample_result.payload)
     self.assertEqual(
         rdfvalue.RDFDatetime.FromMicrosecondsSinceEpoch(results[0].timestamp),
-        rdfvalue.RDFDatetime.FromSecondsSinceEpoch(42),
+        current_time,
     )
 
   def testWritesAndReadsMultipleFlowResultsOfSingleType(self):
@@ -2743,17 +2745,16 @@ def testWritesAndReadsSingleFlowErrorOfSingleType(self):
     sample_error = flows_pb2.FlowError(client_id=client_id, flow_id=flow_id)
     sample_error.payload.Pack(jobs_pb2.ClientSummary(client_id=client_id))
 
-    with test_lib.FakeTime(42):
+    current_time = rdfvalue.RDFDatetime.Now()
+    with test_lib.FakeTime(current_time):
       self.db.WriteFlowErrors([sample_error])
 
     errors = self.db.ReadFlowErrors(client_id, flow_id, 0, 100)
     self.assertLen(errors, 1)
     self.assertEqual(errors[0].payload, sample_error.payload)
     self.assertEqual(
         errors[0].timestamp,
-        rdfvalue.RDFDatetime.FromSecondsSinceEpoch(
-            42
-        ).AsMicrosecondsSinceEpoch(),
+        current_time.AsMicrosecondsSinceEpoch(),
     )
 
   def testWritesAndReadsMultipleFlowErrorsOfSingleType(self):

grr/server/grr_response_server/databases/db_message_handler_test.py

@@ -93,6 +93,7 @@ def testMessageHandlerRequestLeasing(self):
         m.ClearField("timestamp")
       got += l
     self.db.DeleteMessageHandlerRequests(got)
+    self.db.UnregisterMessageHandler()
 
     got.sort(key=lambda req: req.request_id)
     self.assertEqual(requests, got)

grr/server/grr_response_server/databases/registry_init.py

@@ -3,9 +3,11 @@
 
 from grr_response_server.databases import mem
 from grr_response_server.databases import mysql
+from grr_response_server.databases import spanner
 
 # All available databases go into this registry.
 REGISTRY = {
     "InMemoryDB": mem.InMemoryDB,
     "MysqlDB": mysql.MysqlDB,
+    "SpannerDB": spanner.SpannerDB.FromConfig,
 }

grr/server/grr_response_server/databases/spanner.md

@@ -0,0 +1,80 @@
+# GRR on Spanner
+
+When operating GRR you need to decide on a persistence store.
+
+This document describes how to use [Spanner](https://cloud.google.com/spanner)
+as the GRR datastore.
+
+Spanner is a fully managed, mission-critical database service on
+[Google Cloud](https://cloud.google.com) that brings together relational, graph,
+key-value, and search. It offers transactional consistency at global scale,
+automatic, synchronous replication for high availability.
+
+## 1. Google Cloud Resources
+
+Running GRR on Spanner requires that you create and configure a
+[Spanner instance](https://cloud.google.com/spanner/docs/instances) before you
+run GRR.
+
+Furthermore, you also need to create a
+[Google Cloud Storage](https://cloud.google.com/storage)
+[Bucket](https://cloud.google.com/storage/docs/buckets) that
+will serve as the GRR blobstore.
+
+## 2. Google Cloud Spanner Instance
+
+You can follow the instructions in the Google Cloud online documentation to
+[create a Spanner instance](https://cloud.google.com/spanner/docs/create-query-database-console#create-instance).
+
+> [!NOTE] You only need to create the
+> [Spanner instance](https://cloud.google.com/spanner/docs/instances). The
+> GRR [Spanner database](https://cloud.google.com/spanner/docs/databases)
+> and its tables are created by running the provided [spanner_setup.sh](./spanner_setup.sh)
+> script. The script assumes that you use `grr-instance` as the
+> GRR instance name and `grr-database` as the GRR database name. In
+> case you want to use different values then you need to update the
+> [spanner_setup.sh](./spanner_setup.sh) script accordingly.
+> The script assumes that you have the
+> [gcloud](https://cloud.google.com/sdk/docs/install) and the
+> [protoc](https://protobuf.dev/installation/) binaries installed on your machine.
+
+Run the following command to create the GRR database and its tables:
+
+```bash
+export PROJECT_ID=<YOUR_PROJECT_ID_HERE>
+export SPANNER_INSTANCE=grr-instance
+export SPANNER_DATABASE=grr-database
+./spanner_setup.sh
+```
+
+## 3. GRR Configuration
+
+To run GRR on Spanner you need to configure the components settings with  
+the values of the Google Cloud Spanner and the GCS Bucket resources mentioned above.
+
+The snippet below illustrates a sample GRR `server.yaml` configuration.
+
+```bash
+    Database.implementation: SpannerDB
+    Spanner.project: <YOUR_PROJECT_ID_HERE>
+    Spanner.instance: grr-instance
+    Spanner.database: grr-database
+    Blobstore.implementation: GCSBlobStore
+    Blobstore.gcs.project: <YOUR_PROJECT_ID_HERE>
+    Blobstore.gcs.bucket: <YOUR_GCS_BUCKET_NAME_HERE>
+```
+
+> [!NOTE] Make sure you remove all the `Mysql` related configuration items.
+
+## 4. IAM Permissions
+
+This guide assumes that your GRR instance is running on the [Google Kubernetes Engine](https://cloud.google.com/kubernetes-engine) (GKE) 
+and you can leverage [Workload Identity Federation for GKE](https://cloud.google.com/kubernetes-engine/docs/how-to/workload-identity) (WIF).
+
+Using WIF you can assign the required IAM roles using the WIF principal
+`principal://iam.googleapis.com/projects/<PROJECT_NUMBER>/locations/global/workloadIdentityPools/<PROJECT_ID.svc.id.goog/subject/ns/<K8S_NAMESPACE>/sa/<K8S_SERVICE_ACCOUNT>` 
+where `K8S_NAMESPACE` is the value of the Kubernetes Namespace and `K8S_SERVICE_ACCOUNT` is the value Kubernetes Service Account that your GRR Pods are running under.
+
+The two IAM roles that are required are:
+- `roles/spanner.databaseUser` on your Spanner Database and
+- `roles/storage.objectUser` on our GCS Bucket (the GRR Blobstore mentioned above).