PM-1286 Implement authentication for hosted Cassandra

README.md

@@ -114,15 +114,27 @@ If the `CONFIG_FILE` environment variable is not set, the program will fall back
    - `AWS_ACCESS_KEY_ID` - Your AWS Access Key ID.
    - `AWS_SECRET_ACCESS_KEY` - Your AWS Secret Access Key.
 
-4. **AWS Keyspaces Configuration**:
-   - `AWS_KEYSPACE` - Your AWS Keyspace name.
+4. **AWS Keyspaces/Cassandra Configuration**:
+
+   **Mandatory/common env vars:**
+   - `AWS_KEYSPACE` - Your Keyspace name.
+   - `AWS_SSL_CERTIFICATE_PATH` - The path to your SSL certificate.
+
+   **Depending on way of connecting:**
+
+   _Service level connection:_
+   - `CASSANDRA_HOST` - Cassandra host (e.g. cassandra.us-west-2.amazonaws.com).
+   - `CASSANDRA_PORT` - Cassandra port (e.g. 9142).
+   - `CASSANDRA_USERNAME` - Cassandra service user.
+   - `CASSANDRA_PASSWORD` - Cassandra service password.
+
+   _AWS access key / web identity token:_
    - `AWS_REGION` - The AWS region (same as used for S3).
    - `AWS_WEB_IDENTITY_TOKEN_FILE` - AWS web identity token file.
    - `AWS_ROLE_SESSION_NAME` - AWS role session name.
    - `AWS_ROLE_ARN` - AWS role arn.
    - `AWS_ACCESS_KEY_ID` - Your AWS Access Key ID. No need to set if `AWS_WEB_IDENTITY_TOKEN_FILE`, `AWS_ROLE_SESSION_NAME` and `AWS_ROLE_ARN` are set.
    - `AWS_SECRET_ACCESS_KEY` - Your AWS Secret Access Key. No need to set if `AWS_WEB_IDENTITY_TOKEN_FILE`, `AWS_ROLE_SESSION_NAME` and `AWS_ROLE_ARN` are set.
-   - `AWS_SSL_CERTIFICATE_PATH` - The path to your SSL certificate for AWS Keyspaces.
 
 > **Note:** Docker image already includes cert and has `AWS_SSL_CERTIFICATE_PATH` set up, however it can be overriden by providing this env variable to docker.
 

src/delegation_backend/app_config.go

@@ -3,6 +3,7 @@ package delegation_backend
 import (
 	"encoding/json"
 	"os"
+	"strconv"
 
 	logging "github.com/ipfs/go-log/v2"
 )
@@ -71,13 +72,24 @@ func LoadEnv(log logging.EventLogger) AppConfig {
 			}
 		}
 
-		// AWSKeyspace configurations
-		if awsKeyspace := os.Getenv("AWS_KEYSPACE"); awsKeyspace != "" {
-
-			awsRegion := getEnvChecked("AWS_REGION", log)
+		// AWSKeyspace/Cassandra configurations
+		if keyspace := os.Getenv("AWS_KEYSPACE"); keyspace != "" {
 			awsKeyspace := getEnvChecked("AWS_KEYSPACE", log)
 			sslCertificatePath := getEnvChecked("AWS_SSL_CERTIFICATE_PATH", log)
 
+			//service level connection
+			cassandraHost := os.Getenv("CASSANDRA_HOST")
+			cassandraPortStr := os.Getenv("CASSANDRA_PORT")
+			cassandraPort, err := strconv.Atoi(cassandraPortStr)
+			if err != nil {
+				cassandraPort = 9142
+			}
+			cassandraUsername := os.Getenv("CASSANDRA_USERNAME")
+			cassandraPassword := os.Getenv("CASSANDRA_PASSWORD")
+
+			//aws keyspaces connection
+			awsRegion := os.Getenv("AWS_REGION")
+
 			// if webIdentityTokenFile, roleSessionName and roleArn are set,
 			// we are using AWS STS to assume a role and get temporary credentials
 			// if they are not set, we are using AWS IAM user credentials
@@ -90,6 +102,10 @@ func LoadEnv(log logging.EventLogger) AppConfig {
 
 			config.AwsKeyspaces = &AwsKeyspacesConfig{
 				Keyspace:             awsKeyspace,
+				CassandraHost:        cassandraHost,
+				CassandraPort:        cassandraPort,
+				CassandraUsername:    cassandraUsername,
+				CassandraPassword:    cassandraPassword,
 				Region:               awsRegion,
 				AccessKeyId:          accessKeyId,
 				SecretAccessKey:      secretAccessKey,
@@ -150,7 +166,11 @@ type AwsConfig struct {
 
 type AwsKeyspacesConfig struct {
 	Keyspace             string `json:"keyspace"`
-	Region               string `json:"region"`
+	CassandraHost        string `json:"cassandra_host"`
+	CassandraPort        int    `json:"cassandra_port"`
+	CassandraUsername    string `json:"cassandra_username,omitempty"`
+	CassandraPassword    string `json:"cassandra_password,omitempty"`
+	Region               string `json:"region,omitempty"`
 	AccessKeyId          string `json:"access_key_id,omitempty"`
 	SecretAccessKey      string `json:"secret_access_key,omitempty"`
 	WebIdentityTokenFile string `json:"web_identity_token_file,omitempty"`

src/delegation_backend/aws_keyspaces.go

@@ -23,19 +23,71 @@ import (
 
 // InitializeKeyspaceSession creates a new gocql session for Amazon Keyspaces using the provided configuration.
 func InitializeKeyspaceSession(config *AwsKeyspacesConfig) (*gocql.Session, error) {
-	auth := sigv4.NewAwsAuthenticator()
+	var cluster *gocql.ClusterConfig
+
+	var endpoint string
+	if config.CassandraHost == "" {
+		if config.Region == "" {
+			return nil, fmt.Errorf("AWS_REGION is required when CASSANDRA_HOST is not set")
+		}
+		endpoint = "cassandra." + config.Region + ".amazonaws.com"
+	} else {
+		endpoint = config.CassandraHost
+	}
+
+	cluster = gocql.NewCluster(endpoint)
+	cluster.Keyspace = config.Keyspace
+
+	var port int
+	if config.CassandraPort != 0 {
+		port = config.CassandraPort
+	} else {
+		port = 9142
+	}
+	cluster.Port = port
+
+	if config.CassandraUsername != "" && config.CassandraPassword != "" {
+		cluster.Authenticator = gocql.PasswordAuthenticator{
+			Username: config.CassandraUsername,
+			Password: config.CassandraPassword}
+	} else {
+		var err error
+		cluster.Authenticator, err = sigv4Authentication(config)
+		if err != nil {
+			return nil, fmt.Errorf("could not create SigV4 authenticator: %w", err)
+		}
+	}
+
+	cluster.SslOpts = &gocql.SslOptions{
+		CaPath: config.SSLCertificatePath,
+
+		EnableHostVerification: false,
+	}
+
+	cluster.Consistency = gocql.LocalQuorum
+	cluster.DisableInitialHostLookup = false
 
+	session, err := cluster.CreateSession()
+	if err != nil {
+		return nil, fmt.Errorf("could not create Cassandra session: %w", err)
+	}
+
+	return session, nil
+}
+
+func sigv4Authentication(config *AwsKeyspacesConfig) (sigv4.AwsAuthenticator, error) {
+	auth := sigv4.NewAwsAuthenticator()
 	if config.RoleSessionName != "" && config.RoleArn != "" && config.WebIdentityTokenFile != "" {
 		// If role-related env variables are set, use temporary credentials
 		tokenBytes, err := os.ReadFile(config.WebIdentityTokenFile)
 		if err != nil {
-			return nil, fmt.Errorf("error reading web identity token file: %w", err)
+			return auth, fmt.Errorf("error reading web identity token file: %w", err)
 		}
 		webIdentityToken := string(tokenBytes)
 
 		awsSession, err := session.NewSession(&aws.Config{Region: aws.String(config.Region)})
 		if err != nil {
-			return nil, fmt.Errorf("error creating AWS session: %w", err)
+			return auth, fmt.Errorf("error creating AWS session: %w", err)
 		}
 
 		stsSvc := sts.New(awsSession)
@@ -45,7 +97,7 @@ func InitializeKeyspaceSession(config *AwsKeyspacesConfig) (*gocql.Session, erro
 			WebIdentityToken: &webIdentityToken,
 		})
 		if err != nil {
-			return nil, fmt.Errorf("unable to assume role: %w", err)
+			return auth, fmt.Errorf("unable to assume role: %w", err)
 		}
 
 		auth.AccessKeyId = *creds.Credentials.AccessKeyId
@@ -58,28 +110,7 @@ func InitializeKeyspaceSession(config *AwsKeyspacesConfig) (*gocql.Session, erro
 		auth.SecretAccessKey = config.SecretAccessKey
 		auth.Region = config.Region
 	}
-
-	// Create a SigV4 gocql cluster config
-	endpoint := "cassandra." + config.Region + ".amazonaws.com"
-	cluster := gocql.NewCluster(endpoint)
-	cluster.Keyspace = config.Keyspace
-	cluster.Port = 9142
-	cluster.Authenticator = auth
-	cluster.SslOpts = &gocql.SslOptions{
-		CaPath:                 config.SSLCertificatePath,
-		EnableHostVerification: false,
-	}
-
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.DisableInitialHostLookup = false
-
-	// Create a SigV4 gocql session
-	session, err := cluster.CreateSession()
-	if err != nil {
-		return nil, fmt.Errorf("could not create Cassandra session: %w", err)
-	}
-
-	return session, nil
+	return auth, nil
 }
 
 type Submission struct {