diff --git a/.github/workflows/compatibility.yml b/.github/workflows/compatibility.yml index 494f633d77..bec4fe27cb 100644 --- a/.github/workflows/compatibility.yml +++ b/.github/workflows/compatibility.yml @@ -201,6 +201,7 @@ jobs: -e FLOCI_SERVICES_LAMBDA_HOT_RELOAD_ENABLED=true \ -e FLOCI_TLS_ENABLED=true \ -e FLOCI_SERVICES_EC2_MOCK=true \ + -e FLOCI_SERVICES_REDSHIFT_ENDPOINT_HOST=floci \ floci:test-native - name: Wait for floci to be ready diff --git a/compatibility-tests/sdk-test-java/src/test/java/io/github/hectorvent/floci/compat/RedshiftTest.java b/compatibility-tests/sdk-test-java/src/test/java/io/github/hectorvent/floci/compat/RedshiftTest.java index e8380e3554..0d8f044e0b 100644 --- a/compatibility-tests/sdk-test-java/src/test/java/io/github/hectorvent/floci/compat/RedshiftTest.java +++ b/compatibility-tests/sdk-test-java/src/test/java/io/github/hectorvent/floci/compat/RedshiftTest.java @@ -16,6 +16,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; @TestMethodOrder(MethodOrderer.OrderAnnotation.class) public class RedshiftTest { @@ -44,8 +45,13 @@ public void testCreateCluster() throws Exception { Cluster cluster = describeRes.clusters().get(0); assertEquals("test-cluster", cluster.clusterIdentifier()); assertNotNull(cluster.endpoint()); - // No JDBC connection here: the endpoint carries the backing container's own host/port - // (Redshift has no RDS-style auth proxy yet), which isn't reachable from the harness. + String address = cluster.endpoint().address(); + int port = cluster.endpoint().port(); + String jdbcUrl = "jdbc:postgresql://" + address + ":" + port + "/dev"; + try (java.sql.Connection conn = + java.sql.DriverManager.getConnection(jdbcUrl, "admin", "Password123")) { + assertTrue(conn.isValid(5)); + } } @Test diff --git a/docs/services/redshift.md b/docs/services/redshift.md index 0d632298b0..5e1a295b6e 100644 --- a/docs/services/redshift.md +++ b/docs/services/redshift.md @@ -2,9 +2,9 @@ **Protocol:** Query (XML) for the management API **Management Endpoint:** `POST http://localhost:4566/` with `Action=` param -**Data Endpoint:** the `Endpoint` and `Port` returned by `DescribeClusters` (PostgreSQL wire protocol) +**Data Endpoint:** Floci's auth proxy on the `Endpoint` and `Port` returned by `DescribeClusters` (PostgreSQL wire protocol) -Floci emulates Amazon Redshift by managing a real [PostgreSQL](https://www.postgresql.org/) Docker container per cluster behind a Redshift-shaped control plane. Redshift speaks the PostgreSQL wire protocol, so the cluster endpoint returned by `DescribeClusters` works with any standard PostgreSQL driver (`psql`, JDBC, `psycopg`, …). +Floci emulates Amazon Redshift by managing a real [PostgreSQL](https://www.postgresql.org/) Docker container per cluster behind a Redshift-shaped control plane. Each cluster sits behind a lightweight auth proxy on the Floci host, so the endpoint is reachable from outside Docker and the master password is validated at the proxy — a `ModifyCluster` password change takes effect for new connections immediately. Redshift speaks the PostgreSQL wire protocol, so the cluster endpoint returned by `DescribeClusters` works with any standard PostgreSQL driver (`psql`, JDBC, `psycopg`, …). > **Always read the host and port from `DescribeClusters`** rather than assuming a fixed port. PostgreSQL listens on `5432` *inside* the container; the port you connect to is dynamically assigned on the host and returned as `Clusters[0].Endpoint.Port`. Redshift's conventional port is `5439`, but the emulator does not bind it — use whatever `DescribeClusters` reports. @@ -45,6 +45,9 @@ The container has **no persistent volume**: if the physical container survives a | `FLOCI_SERVICES_REDSHIFT_ENABLED` | `true` | Enable or disable Redshift | | `FLOCI_SERVICES_REDSHIFT_IMAGE_VERSION` | `postgres:15-alpine` | PostgreSQL Docker image backing each cluster | | `FLOCI_SERVICES_REDSHIFT_DEFAULT_PORT` | `5439` | Reported Redshift port hint (the real host port is dynamic and comes from `DescribeClusters`) | +| `FLOCI_SERVICES_REDSHIFT_PROXY_BASE_PORT` | `7100` | Lowest host port the per-cluster auth proxies bind | +| `FLOCI_SERVICES_REDSHIFT_PROXY_MAX_PORT` | `7199` | Highest host port the per-cluster auth proxies bind | +| `FLOCI_SERVICES_REDSHIFT_ENDPOINT_HOST` | _(unset)_ | Hostname advertised in `DescribeClusters`; unset resolves from the Docker host | Redshift needs the Docker socket so it can launch PostgreSQL containers. Each cluster's container is published on a dynamically assigned host port, returned by `DescribeClusters`. @@ -139,3 +142,5 @@ print(cluster["Cluster"]["Endpoint"]) - Parameter groups apply no real engine settings; values are stored and echoed back only. - Subnet groups, VPC routing, and security groups are metadata only. - Resize, pause/resume, IAM authentication, snapshot schedules, and cross-region snapshot copy. +- The auth proxy validates only the master user's password. Non-master users pass straight through to PostgreSQL, which remains the authority for their credentials. +- IAM database authentication (`GetClusterCredentials`), and `sslmode=verify-full` against the self-signed proxy certificate. diff --git a/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java b/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java index 1772952a7b..3a56d9cb1e 100644 --- a/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java +++ b/src/main/java/io/github/hectorvent/floci/config/EmulatorConfig.java @@ -1131,6 +1131,17 @@ interface RedshiftServiceConfig { @WithDefault("postgres:15-alpine") String imageVersion(); Optional dockerNetwork(); + + // Port range the per-cluster auth proxies bind on the Floci host. Disjoint from + // every other service's range (RDS uses 7001-7099). + @WithDefault("7100") + int proxyBasePort(); + @WithDefault("7199") + int proxyMaxPort(); + + // Hostname clients use to reach a cluster endpoint. Empty -> resolved from + // DockerHostResolver (falls back to "localhost"). + Optional endpointHost(); } interface RdsServiceConfig { diff --git a/src/main/java/io/github/hectorvent/floci/services/rds/proxy/RdsAuthProxy.java b/src/main/java/io/github/hectorvent/floci/services/rds/proxy/RdsAuthProxy.java index 9f5a542982..a72a261419 100644 --- a/src/main/java/io/github/hectorvent/floci/services/rds/proxy/RdsAuthProxy.java +++ b/src/main/java/io/github/hectorvent/floci/services/rds/proxy/RdsAuthProxy.java @@ -93,9 +93,10 @@ private void acceptLoop() { } private void handleConnection(Socket client) { + Socket backend = null; try { client.setTcpNoDelay(true); - Socket backend = new Socket(backendHost, backendPort); + backend = new Socket(backendHost, backendPort); backend.setTcpNoDelay(true); switch (engine) { @@ -108,7 +109,14 @@ private void handleConnection(Socket client) { } } catch (Exception e) { LOG.debugv("RDS connection error for instance {0}: {1}", instanceId, e.getMessage()); + } finally { + // A handler's success path bridges then closes both sockets; every other path + // (early return on a bare probe, auth failure, thrown IOException) can leave the + // backend DB connection open. Closing here is idempotent. closeQuietly(client); + if (backend != null) { + closeQuietly(backend); + } } } diff --git a/src/main/java/io/github/hectorvent/floci/services/redshift/RedshiftService.java b/src/main/java/io/github/hectorvent/floci/services/redshift/RedshiftService.java index 40cc75c875..997d381b65 100644 --- a/src/main/java/io/github/hectorvent/floci/services/redshift/RedshiftService.java +++ b/src/main/java/io/github/hectorvent/floci/services/redshift/RedshiftService.java @@ -4,7 +4,10 @@ import io.github.hectorvent.floci.core.common.AwsArnUtils; import io.github.hectorvent.floci.core.common.AwsException; import io.github.hectorvent.floci.core.common.RegionResolver; +import io.github.hectorvent.floci.core.common.docker.DockerHostResolver; import io.github.hectorvent.floci.core.storage.AccountAwareStorageBackend; +import io.github.hectorvent.floci.services.rds.proxy.RdsAuthProxy; +import io.github.hectorvent.floci.services.redshift.proxy.RedshiftProxyManager; import io.github.hectorvent.floci.core.storage.StorageFactory; import io.github.hectorvent.floci.services.redshift.container.RedshiftContainerHandle; import io.github.hectorvent.floci.services.redshift.container.RedshiftContainerManager; @@ -31,6 +34,8 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; @ApplicationScoped @@ -44,10 +49,15 @@ public class RedshiftService { private final RedshiftContainerManager containerManager; private final EmulatorConfig config; private final RegionResolver regionResolver; + private final RedshiftProxyManager proxyManager; + private final DockerHostResolver dockerHostResolver; + // Proxy ports currently handed out, so allocateProxyPort never double-assigns within this JVM. + private final Set usedPorts = ConcurrentHashMap.newKeySet(); @Inject public RedshiftService(StorageFactory storageFactory, RedshiftContainerManager containerManager, - EmulatorConfig config, RegionResolver regionResolver) { + EmulatorConfig config, RegionResolver regionResolver, + RedshiftProxyManager proxyManager, DockerHostResolver dockerHostResolver) { this.clusters = storageFactory.create("redshift", "redshift-clusters.json", new TypeReference>() {}); this.snapshots = storageFactory.create("redshift", "redshift-snapshots.json", new TypeReference>() {}); this.parameterGroups = storageFactory.create("redshift", "redshift-parameter-groups.json", new TypeReference>() {}); @@ -55,6 +65,8 @@ public RedshiftService(StorageFactory storageFactory, RedshiftContainerManager c this.containerManager = containerManager; this.config = config; this.regionResolver = regionResolver; + this.proxyManager = proxyManager; + this.dockerHostResolver = dockerHostResolver; } // Recreate Docker containers for persisted clusters on app restart (across every account, not just default) @@ -73,14 +85,36 @@ void onStart(@Observes StartupEvent event) { LOG.infov("Recovering container for persisted cluster: {0}", cluster.getClusterIdentifier()); RedshiftContainerHandle handle = containerManager.adoptOrStart( entry.accountId(), cluster.getClusterIdentifier(), cluster.getMasterUsername(), password); - Endpoint endpoint = new Endpoint(); - endpoint.setAddress(handle.getHost()); - endpoint.setPort(handle.getPort()); + + // A cluster persisted before the auth proxy existed has proxyPort == 0. Allocate one + // now; its endpoint changes exactly once after this upgrade. Existing clusters keep + // their stored port so the endpoint is stable across restarts. + int proxyPort = cluster.getProxyPort() > 0 ? cluster.getProxyPort() : allocateProxyPort(); + usedPorts.add(proxyPort); + Endpoint endpoint = proxyEndpoint(proxyPort); + cluster.setProxyPort(proxyPort); + proxyManager.startProxy( + relayKey(entry.accountId(), cluster.getClusterIdentifier()), proxyPort, + handle.getHost(), handle.getPort(), endpoint.getAddress(), + cluster.getMasterUsername(), password, CLUSTER_DB_NAME, + passwordValidatorFor(entry.accountId(), cluster.getClusterIdentifier())); + cluster.setContainerHost(handle.getHost()); + cluster.setContainerPort(handle.getPort()); cluster.setEndpoint(endpoint); clusters.putForAccount(entry.accountId(), entry.key(), cluster); } catch (Exception e) { LOG.warnv(e, "Failed to recover container for cluster {0}, marking as unavailable", cluster.getClusterIdentifier()); + try { + proxyManager.stopProxy(relayKey(entry.accountId(), cluster.getClusterIdentifier())); + } catch (Exception ex) { + LOG.warnv(ex, "Failed to stop proxy during recovery rollback for cluster {0}", cluster.getClusterIdentifier()); + } + try { + containerManager.stop(entry.accountId(), cluster.getClusterIdentifier()); + } catch (Exception ex) { + LOG.warnv(ex, "Failed to stop container during recovery rollback for cluster {0}", cluster.getClusterIdentifier()); + } cluster.setClusterStatus("unavailable"); clusters.putForAccount(entry.accountId(), entry.key(), cluster); } @@ -92,7 +126,9 @@ public Cluster createCluster(String identifier, String nodeType, String username return createCluster(identifier, nodeType, username, password, null, List.of()); } - public Cluster createCluster(String identifier, String nodeType, String username, String password, + // synchronized like modify/reboot: the container + proxy + port steps must not + // interleave with another admin call on the same cluster. + public synchronized Cluster createCluster(String identifier, String nodeType, String username, String password, String clusterSubnetGroupName, List vpcSecurityGroupIds) { if (clusters.get(identifier).isPresent()) { throw new AwsException("ClusterAlreadyExists", "Cluster " + identifier + " already exists", 400); @@ -109,16 +145,44 @@ public Cluster createCluster(String identifier, String nodeType, String username clusters.put(identifier, cluster); clusters.flush(); - // Start container + // Start container, then front it with an auth proxy so the advertised endpoint is + // reachable from outside the Docker network. + // Hoisted out of the try so a failure after allocateProxyPort() still returns the port. + int proxyPort = -1; try { - RedshiftContainerHandle handle = containerManager.start(clusters.accountId(), identifier, username, password); - Endpoint endpoint = new Endpoint(); - endpoint.setAddress(handle.getHost()); - endpoint.setPort(handle.getPort()); + String accountId = clusters.accountId(); + RedshiftContainerHandle handle = containerManager.start(accountId, identifier, username, password); + proxyPort = allocateProxyPort(); + Endpoint endpoint = proxyEndpoint(proxyPort); + cluster.setProxyPort(proxyPort); + proxyManager.startProxy(relayKey(accountId, identifier), proxyPort, + handle.getHost(), handle.getPort(), endpoint.getAddress(), + username, password, CLUSTER_DB_NAME, + passwordValidatorFor(accountId, identifier)); + cluster.setContainerHost(handle.getHost()); + cluster.setContainerPort(handle.getPort()); cluster.setEndpoint(endpoint); cluster.setClusterStatus("available"); + } catch (AwsException e) { + boolean proxyStopped = stopProxyAndReleasePortSafely(identifier, proxyPort); + try { containerManager.stop(clusters.accountId(), identifier); } catch (Exception ex) { LOG.warnv(ex, "Failed to stop container during rollback of cluster {0}", identifier); } + if (proxyStopped) { + clusters.delete(identifier); + } else { + cluster.setClusterStatus("failed"); + clusters.put(identifier, cluster); + } + clusters.flush(); + throw e; } catch (Exception e) { - cluster.setClusterStatus("failed"); + boolean proxyStopped = stopProxyAndReleasePortSafely(identifier, proxyPort); + try { containerManager.stop(clusters.accountId(), identifier); } catch (Exception ex) { LOG.warnv(ex, "Failed to stop container during rollback of cluster {0}", identifier); } + if (proxyStopped) { + clusters.delete(identifier); + } else { + cluster.setClusterStatus("failed"); + clusters.put(identifier, cluster); + } clusters.flush(); throw new AwsException("InternalFailure", "Failed to start container: " + e.getMessage(), 500); } @@ -139,13 +203,20 @@ public List describeClusters(String identifier) { return clusters.scan(k -> true); } - public Cluster deleteCluster(String identifier) { + public synchronized Cluster deleteCluster(String identifier) { Optional clusterOpt = clusters.get(identifier); if (clusterOpt.isEmpty()) { throw new AwsException("ClusterNotFound", "Cluster " + identifier + " not found", 404); } Cluster cluster = clusterOpt.get(); - + + // Tear down the auth proxy and return its port before stopping the container. + // If it fails, abort deletion so the metadata remains and the user can retry. + if (!stopProxyAndReleasePortSafely(identifier, cluster.getProxyPort())) { + throw new AwsException("InternalFailure", + "Failed to stop auth proxy; cluster " + identifier + " was not deleted", 500); + } + containerManager.stop(clusters.accountId(), identifier); clusters.delete(identifier); clusters.flush(); @@ -166,6 +237,9 @@ public synchronized Cluster modifyCluster(String clusterIdentifier, String nodeT containerManager.alterUserPassword(clusters.accountId(), clusterIdentifier, cluster.getMasterUsername(), masterUserPassword); cluster.setMasterPassword(masterUserPassword); + // Keep the proxy's password check in sync so new connections use the new secret. + proxyManager.updateMasterPassword( + relayKey(clusters.accountId(), clusterIdentifier), masterUserPassword); } // NodeType only updates metadata — it does not resize the underlying Postgres container @@ -201,34 +275,74 @@ public synchronized Cluster rebootCluster(String clusterIdentifier) { throw new AwsException("InternalFailure", "Failed to prepare reboot dump file: " + e.getMessage(), 500); } + // Hoisted so a failure after the proxy is (re)started still tears it down and + // returns the port, matching createCluster/restoreFromClusterSnapshot rollback. + int proxyPort = cluster.getProxyPort() > 0 ? cluster.getProxyPort() : -1; + boolean originalTornDown = false; // original proxy + container already stopped + boolean rebooted = false; try { - containerManager.takeSnapshot(clusters.accountId(), clusterIdentifier, cluster.getMasterUsername(), tempDump); - containerManager.stop(clusters.accountId(), clusterIdentifier); + String accountId = clusters.accountId(); + String key = relayKey(accountId, clusterIdentifier); + + containerManager.takeSnapshot(accountId, clusterIdentifier, cluster.getMasterUsername(), tempDump); + proxyManager.stopProxy(key); + containerManager.stop(accountId, clusterIdentifier); + originalTornDown = true; String password = cluster.getMasterPassword() != null ? cluster.getMasterPassword() : "admin"; RedshiftContainerHandle handle = containerManager.start( - clusters.accountId(), clusterIdentifier, cluster.getMasterUsername(), password); - Endpoint endpoint = new Endpoint(); - endpoint.setAddress(handle.getHost()); - endpoint.setPort(handle.getPort()); - cluster.setEndpoint(endpoint); + accountId, clusterIdentifier, cluster.getMasterUsername(), password); - containerManager.restoreSnapshot(clusters.accountId(), clusterIdentifier, cluster.getMasterUsername(), tempDump); + // Reuse the stored proxy port so the advertised endpoint is unchanged by a reboot. + if (proxyPort < 0) { + proxyPort = allocateProxyPort(); + } + usedPorts.add(proxyPort); + Endpoint endpoint = proxyEndpoint(proxyPort); + cluster.setProxyPort(proxyPort); + proxyManager.startProxy(key, proxyPort, handle.getHost(), handle.getPort(), + endpoint.getAddress(), cluster.getMasterUsername(), password, CLUSTER_DB_NAME, + passwordValidatorFor(accountId, clusterIdentifier)); + cluster.setContainerHost(handle.getHost()); + cluster.setContainerPort(handle.getPort()); + cluster.setEndpoint(endpoint); + containerManager.restoreSnapshot(accountId, clusterIdentifier, cluster.getMasterUsername(), tempDump); cluster.setClusterStatus("available"); + rebooted = true; } catch (AwsException e) { - cluster.setClusterStatus("failed"); + rollbackReboot(clusterIdentifier, originalTornDown); + if (originalTornDown) { + cluster.setClusterStatus("failed"); + } clusters.flush(); throw e; } catch (Exception e) { - cluster.setClusterStatus("failed"); + rollbackReboot(clusterIdentifier, originalTornDown); + if (originalTornDown) { + cluster.setClusterStatus("failed"); + } clusters.flush(); throw new AwsException("InternalFailure", "Failed to reboot cluster " + clusterIdentifier + ": " + e.getMessage(), 500); } finally { - try { - Files.deleteIfExists(tempDump); - } catch (IOException ignored) { - // best-effort cleanup of a temp file + if (rebooted) { + try { + Files.deleteIfExists(tempDump); + } catch (IOException ex) { + LOG.warnv(ex, "Failed to clean up temporary dump file {0} after rebooting cluster {1}", tempDump, clusterIdentifier); + } + } else { + if (originalTornDown) { + // Once the original container is torn down it holds no volume, so this dump can + // be the only surviving copy of the cluster's data — keep it for manual recovery. + LOG.warnv("Reboot of cluster {0} did not complete; retained pre-reboot data dump at {1}", + clusterIdentifier, tempDump); + } else { + try { + Files.deleteIfExists(tempDump); + } catch (IOException ignored) { + } + } } } @@ -366,7 +480,7 @@ public Cluster restoreFromClusterSnapshot(String clusterIdentifier, String snaps return restoreFromClusterSnapshot(clusterIdentifier, snapshotIdentifier, null); } - public Cluster restoreFromClusterSnapshot(String clusterIdentifier, String snapshotIdentifier, String nodeType) { + public synchronized Cluster restoreFromClusterSnapshot(String clusterIdentifier, String snapshotIdentifier, String nodeType) { if (clusters.get(clusterIdentifier).isPresent()) { throw new AwsException("ClusterAlreadyExists", "Cluster " + clusterIdentifier + " already exists", 400); } @@ -407,11 +521,20 @@ public Cluster restoreFromClusterSnapshot(String clusterIdentifier, String snaps clusters.put(clusterIdentifier, cluster); clusters.flush(); + // Hoisted out of the try so a failure after allocateProxyPort() still returns the port. + int proxyPort = -1; try { - RedshiftContainerHandle handle = containerManager.start(clusters.accountId(), clusterIdentifier, username, password); - Endpoint endpoint = new Endpoint(); - endpoint.setAddress(handle.getHost()); - endpoint.setPort(handle.getPort()); + String accountId = clusters.accountId(); + RedshiftContainerHandle handle = containerManager.start(accountId, clusterIdentifier, username, password); + proxyPort = allocateProxyPort(); + Endpoint endpoint = proxyEndpoint(proxyPort); + cluster.setProxyPort(proxyPort); + proxyManager.startProxy(relayKey(accountId, clusterIdentifier), proxyPort, + handle.getHost(), handle.getPort(), endpoint.getAddress(), + username, password, CLUSTER_DB_NAME, + passwordValidatorFor(accountId, clusterIdentifier)); + cluster.setContainerHost(handle.getHost()); + cluster.setContainerPort(handle.getPort()); cluster.setEndpoint(endpoint); if (hasDump) { @@ -420,11 +543,25 @@ public Cluster restoreFromClusterSnapshot(String clusterIdentifier, String snaps cluster.setClusterStatus("available"); } catch (AwsException e) { - cluster.setClusterStatus("failed"); + boolean proxyStopped = stopProxyAndReleasePortSafely(clusterIdentifier, proxyPort); + try { containerManager.stop(clusters.accountId(), clusterIdentifier); } catch (Exception ex) { LOG.warnv(ex, "Failed to stop container during rollback of cluster {0}", clusterIdentifier); } + if (proxyStopped) { + clusters.delete(clusterIdentifier); + } else { + cluster.setClusterStatus("failed"); + clusters.put(clusterIdentifier, cluster); + } clusters.flush(); throw e; } catch (Exception e) { - cluster.setClusterStatus("failed"); + boolean proxyStopped = stopProxyAndReleasePortSafely(clusterIdentifier, proxyPort); + try { containerManager.stop(clusters.accountId(), clusterIdentifier); } catch (Exception ex) { LOG.warnv(ex, "Failed to stop container during rollback of cluster {0}", clusterIdentifier); } + if (proxyStopped) { + clusters.delete(clusterIdentifier); + } else { + cluster.setClusterStatus("failed"); + clusters.put(clusterIdentifier, cluster); + } clusters.flush(); throw new AwsException("InternalFailure", "Failed to restore cluster from snapshot: " + e.getMessage(), 500); } @@ -725,4 +862,85 @@ yield new TagHandle(group.getTags(), updated -> { "Tagging for resource type '" + type + "' is not supported: " + resourceName, 400); }; } + + // ── Proxy Helpers (shared with modify/reboot/restore) ──────────────────── + + private static final String CLUSTER_DB_NAME = "dev"; + + private int allocateProxyPort() { + int base = config.services().redshift().proxyBasePort(); + int max = config.services().redshift().proxyMaxPort(); + for (int port = base; port <= max; port++) { + if (usedPorts.add(port)) { + return port; + } + } + throw new AwsException("InsufficientClusterCapacity", + "No available Redshift proxy ports in range " + base + "-" + max, 503); + } + + private boolean stopProxyAndReleasePortSafely(String identifier, int proxyPort) { + boolean proxyStopped = false; + try { + proxyManager.stopProxy(relayKey(clusters.accountId(), identifier)); + proxyStopped = true; + } catch (Exception ex) { + LOG.warnv(ex, "Failed to stop proxy for cluster {0}; leaking proxy port {1} to prevent reallocation", identifier, proxyPort); + } + if (proxyStopped) { + releaseProxyPort(proxyPort); + } + return proxyStopped; + } + + /** + * Undo a failed reboot. If {@code originalTornDown} is false the reboot failed before + * the original proxy + container were stopped, so the original data-bearing container + * is still running and nothing must be touched. Once it is true the original is gone: + * tear down the (replacement's) proxy and return its port, and remove any container + * running under the cluster's name — {@code containerManager.stop} works by name, so + * this also cleans a replacement that {@code containerManager.start} created before + * throwing (e.g. its readiness check timed out). The pre-reboot data dump is kept by + * the caller. + */ + private void rollbackReboot(String identifier, boolean originalTornDown) { + if (!originalTornDown) { + return; + } + try { + proxyManager.stopProxy(relayKey(clusters.accountId(), identifier)); + } catch (Exception ex) { + LOG.warnv(ex, "Failed to stop proxy during reboot rollback for cluster {0}", identifier); + } + try { + containerManager.stop(clusters.accountId(), identifier); + } catch (Exception ex) { + LOG.warnv(ex, "Failed to stop replacement container during rollback of reboot for cluster {0}", identifier); + } + } + + private void releaseProxyPort(int port) { + if (port > 0) { + usedPorts.remove(port); + } + } + + private Endpoint proxyEndpoint(int proxyPort) { + String host = config.services().redshift().endpointHost() + .filter(h -> !h.isBlank()) + .orElseGet(dockerHostResolver::resolve); + return new Endpoint(host, proxyPort); + } + + private String relayKey(String accountId, String clusterIdentifier) { + return accountId + ":" + clusterIdentifier; + } + + // Validates the master password at the proxy against current cluster state, so a + // ModifyCluster password change is reflected for new connections without a proxy restart. + private RdsAuthProxy.PasswordValidator passwordValidatorFor(String accountId, String clusterIdentifier) { + return (user, password) -> clusters.getForAccount(accountId, clusterIdentifier) + .map(c -> user.equals(c.getMasterUsername()) && password.equals(c.getMasterPassword())) + .orElse(false); + } } diff --git a/src/main/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManager.java b/src/main/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManager.java index 002c0059e8..215d927975 100644 --- a/src/main/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManager.java +++ b/src/main/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManager.java @@ -99,11 +99,7 @@ public RedshiftContainerHandle start(String accountId, String clusterIdentifier, LOG.warnv("Failed to stream logs for {0}", containerName); } - try { - Thread.sleep(3000); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + waitForReady(containerName, info.containerId(), masterUsername, "dev"); containers.put(containerKey(accountId, clusterIdentifier), handle); return handle; @@ -143,6 +139,8 @@ public RedshiftContainerHandle adoptOrStart(String accountId, String clusterIden LOG.warnv("Failed to stream logs for {0}", containerName); } + waitForReady(containerName, info.containerId(), masterUsername, "dev"); + containers.put(containerKey(accountId, clusterIdentifier), handle); return handle; } @@ -381,6 +379,41 @@ private byte[] buildSingleFileTar(String filename, byte[] content, int mode) thr } return bos.toByteArray(); } + private void waitForReady(String containerName, String containerId, String username, String dbName) { + String effectiveUser = (username != null && !username.isBlank()) ? username : "postgres"; + String[] cmd = { + "psql", + "-h", "127.0.0.1", + "-v", "ON_ERROR_STOP=1", + "-U", effectiveUser, + "-d", dbName, + "-c", "SELECT 1" + }; + execUntilSuccess(containerName, containerId, cmd, "PostgreSQL readiness check"); + } + + private void execUntilSuccess(String containerName, String containerId, String[] cmd, String description) { + String lastOutput = ""; + for (int attempt = 1; attempt <= 60; attempt++) { + try { + ExecResult result = execInContainer(containerId, cmd, 5); + lastOutput = result.stderr(); + if (result.exitCode() == 0) { + LOG.infov("Initialized {0} in Redshift container {1}", description, containerName); + return; + } + } catch (Exception e) { + lastOutput = e.getMessage(); + } + try { + java.util.concurrent.TimeUnit.SECONDS.sleep(1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted initializing " + description + " in " + containerName, e); + } + } + throw new IllegalStateException("Timed out initializing " + description + " in " + containerName + ": " + lastOutput); + } public record ExecResult(long exitCode, String stdout, String stderr) {} } diff --git a/src/main/java/io/github/hectorvent/floci/services/redshift/model/Cluster.java b/src/main/java/io/github/hectorvent/floci/services/redshift/model/Cluster.java index cd518fabbe..24c2107b5d 100644 --- a/src/main/java/io/github/hectorvent/floci/services/redshift/model/Cluster.java +++ b/src/main/java/io/github/hectorvent/floci/services/redshift/model/Cluster.java @@ -37,6 +37,23 @@ public class Cluster { public List getVpcSecurityGroupIds() { return vpcSecurityGroupIds; } public void setVpcSecurityGroupIds(List vpcSecurityGroupIds) { this.vpcSecurityGroupIds = vpcSecurityGroupIds; } + // Real backend address of this cluster's PostgreSQL container. `endpoint` now points at the + // auth proxy, not the container, so the container address is kept here for proxy wiring and + // for restarting the proxy after a reboot or an adopt-on-startup. + private String containerHost = null; + private int containerPort = 0; + + // Host port the cluster's auth proxy binds. Kept stable across reboot and adopt so the + // advertised endpoint does not change. + private int proxyPort = 0; + + public String getContainerHost() { return containerHost; } + public void setContainerHost(String containerHost) { this.containerHost = containerHost; } + public int getContainerPort() { return containerPort; } + public void setContainerPort(int containerPort) { this.containerPort = containerPort; } + public int getProxyPort() { return proxyPort; } + public void setProxyPort(int proxyPort) { this.proxyPort = proxyPort; } + private Map tags = new LinkedHashMap<>(); public Map getTags() { return tags; } diff --git a/src/main/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftAuthProxy.java b/src/main/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftAuthProxy.java new file mode 100644 index 0000000000..4de297185e --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftAuthProxy.java @@ -0,0 +1,164 @@ +package io.github.hectorvent.floci.services.redshift.proxy; + +import io.github.hectorvent.floci.services.rds.proxy.PostgresProtocolHandler; +import io.github.hectorvent.floci.services.rds.proxy.RdsAuthProxy; +import io.github.hectorvent.floci.services.rds.proxy.RdsProxyTlsCertificates; +import io.github.hectorvent.floci.services.rds.proxy.RdsSigV4Validator; +import org.jboss.logging.Logger; + +import java.io.IOException; +import java.net.BindException; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; + +/** + * TCP auth proxy for a single Redshift cluster's backing PostgreSQL container. + * Redshift speaks the PostgreSQL wire protocol, so the RDS PostgreSQL protocol + * handler is reused verbatim for the auth intercept; client and backend are then + * bridged transparently. IAM auth is never enabled for Redshift, so the SigV4 + * validator is carried only to satisfy the shared handler signature. + */ +public class RedshiftAuthProxy { + + private static final Logger LOG = Logger.getLogger(RedshiftAuthProxy.class); + + private final String clusterKey; + private final String backendHost; + private final int backendPort; + private final String masterUsername; + private volatile String masterPassword; + private final String dbName; + private final RdsSigV4Validator sigV4; + private final RdsProxyTlsCertificates tlsCertificates; + private final RdsAuthProxy.PasswordValidator passwordValidator; + + private volatile boolean running; + private ServerSocket serverSocket; + + public RedshiftAuthProxy(String clusterKey, String backendHost, int backendPort, + String masterUsername, String masterPassword, String dbName, + RdsSigV4Validator sigV4, RdsProxyTlsCertificates tlsCertificates, + RdsAuthProxy.PasswordValidator passwordValidator) { + this.clusterKey = clusterKey; + this.backendHost = backendHost; + this.backendPort = backendPort; + this.masterUsername = masterUsername; + this.masterPassword = masterPassword; + this.dbName = dbName; + this.sigV4 = sigV4; + this.tlsCertificates = tlsCertificates; + this.passwordValidator = passwordValidator; + } + + public void start(int proxyPort) throws IOException { + serverSocket = bindListener(proxyPort); + running = true; + Thread.ofVirtual().name("redshift-proxy-accept-" + clusterKey).start(this::acceptLoop); + LOG.infov("Redshift proxy started for cluster {0} on port {1} -> {2}:{3}", + clusterKey, String.valueOf(proxyPort), backendHost, String.valueOf(backendPort)); + } + + /** + * Bind the listener, retrying briefly on {@link BindException}. A reboot keeps the + * cluster's proxy port fixed so the advertised endpoint is stable, which means the + * old listener is closed and the same port rebound milliseconds later; under load + * the kernel may not have released it yet and {@code SO_REUSEADDR} does not help + * while the previous accept loop is still tearing down. ~1s of retry absorbs that + * window; any other {@link IOException} propagates immediately. + */ + private ServerSocket bindListener(int proxyPort) throws IOException { + BindException lastFailure = null; + for (int attempt = 0; attempt < 40; attempt++) { + ServerSocket socket = new ServerSocket(); + try { + socket.setReuseAddress(true); + socket.bind(new InetSocketAddress(proxyPort)); + return socket; + } catch (BindException e) { + closeQuietly(socket); + lastFailure = e; + } catch (IOException e) { + closeQuietly(socket); + throw e; + } + try { + Thread.sleep(25); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while binding Redshift proxy port " + proxyPort, e); + } + } + throw lastFailure; + } + + private static void closeQuietly(ServerSocket s) { + try { + s.close(); + } catch (IOException e) { + LOG.debugv(e, "Error closing a discarded Redshift proxy listener socket"); + } + } + + /** Swap the master-password snapshot after a rotation; new connections authenticate against it. */ + public void updateMasterPassword(String newPassword) { + this.masterPassword = newPassword; + } + + public void stop() { + running = false; + try { + if (serverSocket != null) { + serverSocket.close(); + } + } catch (IOException e) { + LOG.warnv(e, "Error closing Redshift proxy server socket for cluster {0}", clusterKey); + throw new RuntimeException("Failed to stop Redshift proxy for cluster " + clusterKey, e); + } + } + + private void acceptLoop() { + while (running) { + try { + Socket client = serverSocket.accept(); + Thread.ofVirtual().name("redshift-proxy-conn-" + clusterKey) + .start(() -> handleConnection(client)); + } catch (IOException e) { + if (running) { + LOG.warnv("Accept error for Redshift cluster {0}: {1}", clusterKey, e.getMessage()); + } + } + } + } + + private void handleConnection(Socket client) { + Socket backend = null; + try { + client.setTcpNoDelay(true); + backend = new Socket(backendHost, backendPort); + backend.setTcpNoDelay(true); + // iamEnabled = false: the SigV4 branch inside handleAuth is never taken. + PostgresProtocolHandler.handleAuth( + client, backend, masterUsername, masterPassword, dbName, + false, sigV4, tlsCertificates, passwordValidator::validate); + } catch (Exception e) { + LOG.debugv("Redshift connection error for cluster {0}: {1}", clusterKey, e.getMessage()); + } finally { + // handleAuth's success path bridges then closes both sockets; every other path + // (early return on a bare probe, auth failure, thrown IOException) can leave the + // backend connection to Postgres open. Closing here is idempotent. + closeQuietly(client); + if (backend != null) { + closeQuietly(backend); + } + } + } + + private static void closeQuietly(Socket s) { + try { + s.close(); + } catch (IOException e) { + LOG.debugv(e, "Error closing Redshift proxy client socket"); + } + } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftProxyManager.java b/src/main/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftProxyManager.java new file mode 100644 index 0000000000..a51b8c7ae2 --- /dev/null +++ b/src/main/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftProxyManager.java @@ -0,0 +1,134 @@ +package io.github.hectorvent.floci.services.redshift.proxy; + +import io.github.hectorvent.floci.services.rds.proxy.RdsAuthProxy; +import io.github.hectorvent.floci.services.rds.proxy.RdsProxyTlsCertificates; +import io.github.hectorvent.floci.services.rds.proxy.RdsSigV4Validator; +import io.quarkus.runtime.ShutdownEvent; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.enterprise.event.Observes; +import jakarta.inject.Inject; +import org.jboss.logging.Logger; + +import java.io.IOException; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Registry of all active Redshift auth proxies. One proxy per cluster, keyed by + * the relay key "{accountId}:{clusterId}". Mirrors RdsProxyManager; Redshift is + * always PostgreSQL and never IAM-enabled, so those parameters are dropped. + */ +@ApplicationScoped +public class RedshiftProxyManager { + + private static final Logger LOG = Logger.getLogger(RedshiftProxyManager.class); + + private final RdsSigV4Validator sigV4Validator; + private final RdsProxyTlsCertificates tlsCertificates; + private final ConcurrentHashMap proxies = new ConcurrentHashMap<>(); + /** + * Proxies whose listener could not be closed during a failed start or replace. The + * reference is kept (not discarded) so a later {@link #stopProxy}/{@link #stopAll} + * can retry the close; until one succeeds, {@code stopProxy} keeps throwing so the + * caller leaves the port reserved rather than handing it out to another cluster. + */ + private final ConcurrentHashMap unclosableProxies = new ConcurrentHashMap<>(); + + @Inject + public RedshiftProxyManager(RdsSigV4Validator sigV4Validator, RdsProxyTlsCertificates tlsCertificates) { + this.sigV4Validator = sigV4Validator; + this.tlsCertificates = tlsCertificates; + } + + public synchronized void startProxy(String relayKey, int proxyPort, + String backendHost, int backendPort, String advertisedHost, + String masterUsername, String masterPassword, String dbName, + RdsAuthProxy.PasswordValidator passwordValidator) { + // A prior unclosable entry for this key is left in place: its listener may still + // be bound, and only a successful stop (never a fresh start) may drop it. + // Make sure the self-signed proxy certificate covers the host clients will connect to, + // so sslmode=prefer/require handshakes succeed. + tlsCertificates.ensureHost(advertisedHost); + RedshiftAuthProxy proxy = new RedshiftAuthProxy( + relayKey, backendHost, backendPort, masterUsername, masterPassword, dbName, + sigV4Validator, tlsCertificates, passwordValidator); + try { + proxy.start(proxyPort); + } catch (IOException | RuntimeException e) { + RuntimeException failure = new RuntimeException( + "Failed to start Redshift proxy for cluster " + relayKey + " on port " + proxyPort, e); + cleanupFailedStart(relayKey, proxy, failure); + throw failure; + } + RedshiftAuthProxy previous = proxies.put(relayKey, proxy); + if (previous != null) { + try { + previous.stop(); + } catch (RuntimeException e) { + proxies.put(relayKey, previous); + RuntimeException failure = new RuntimeException( + "Failed to replace Redshift proxy for cluster " + relayKey, e); + cleanupFailedStart(relayKey, proxy, failure); + throw failure; + } + } + } + + public synchronized void updateMasterPassword(String relayKey, String newPassword) { + RedshiftAuthProxy proxy = proxies.get(relayKey); + if (proxy != null) { + proxy.updateMasterPassword(newPassword); + LOG.infov("Updated Redshift proxy master password for cluster {0}", relayKey); + } + } + + public synchronized void stopProxy(String relayKey) { + // Retry any listener a previous failed start/replace could not close. If it still + // cannot be closed this throws, and the entry stays for the next retry. + RedshiftAuthProxy unclosable = unclosableProxies.get(relayKey); + if (unclosable != null) { + unclosable.stop(); + unclosableProxies.remove(relayKey); + LOG.infov("Recovered previously unclosable Redshift proxy for cluster {0}", relayKey); + } + RedshiftAuthProxy proxy = proxies.get(relayKey); + if (proxy != null) { + proxy.stop(); + proxies.remove(relayKey); + LOG.infov("Stopped Redshift proxy for cluster {0}", relayKey); + } + } + + public synchronized void stopAll() { + proxies.forEach((relayKey, proxy) -> { + try { + proxy.stop(); + proxies.remove(relayKey, proxy); + } catch (RuntimeException e) { + LOG.warnv(e, "Failed to stop Redshift proxy for cluster {0} during shutdown", relayKey); + } + }); + unclosableProxies.forEach((relayKey, proxy) -> { + try { + proxy.stop(); + unclosableProxies.remove(relayKey, proxy); + } catch (RuntimeException e) { + LOG.warnv(e, "Failed to close leaked Redshift proxy for cluster {0} during shutdown", relayKey); + } + }); + LOG.info("Stopped all Redshift proxies"); + } + + void onShutdown(@Observes ShutdownEvent event) { + stopAll(); + } + + private void cleanupFailedStart(String relayKey, RedshiftAuthProxy proxy, RuntimeException failure) { + try { + proxy.stop(); + } catch (RuntimeException cleanupFailure) { + // Keep the listener reachable so stopProxy/stopAll can retry closing it later. + unclosableProxies.put(relayKey, proxy); + failure.addSuppressed(cleanupFailure); + } + } +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index e232b035a8..96cc8ace1b 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -318,6 +318,9 @@ floci: enabled: true default-port: 5439 image-version: postgres:15-alpine + proxy-base-port: 7100 + proxy-max-port: 7199 + # endpoint-host: localhost # Hostname clients use; empty -> DockerHostResolver rds: enabled: true mock: false # FLOCI_SERVICES_RDS_MOCK — true = metadata only, no Docker (useful for CI) diff --git a/src/test/java/io/github/hectorvent/floci/config/RedshiftProxyConfigTest.java b/src/test/java/io/github/hectorvent/floci/config/RedshiftProxyConfigTest.java new file mode 100644 index 0000000000..2bb3a8c0c3 --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/config/RedshiftProxyConfigTest.java @@ -0,0 +1,27 @@ +package io.github.hectorvent.floci.config; + +import io.quarkus.test.junit.QuarkusTest; +import jakarta.inject.Inject; +import org.junit.jupiter.api.Test; + +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +@QuarkusTest +class RedshiftProxyConfigTest { + + @Inject + EmulatorConfig config; + + @Test + void redshiftProxyPortRangeHasDefaults() { + assertEquals(7100, config.services().redshift().proxyBasePort()); + assertEquals(7199, config.services().redshift().proxyMaxPort()); + } + + @Test + void redshiftEndpointHostDefaultsToEmpty() { + assertEquals(Optional.empty(), config.services().redshift().endpointHost()); + } +} diff --git a/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftOperationsTest.java b/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftOperationsTest.java index 8c969c2b54..c6a15db510 100644 --- a/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftOperationsTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftOperationsTest.java @@ -9,6 +9,10 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestMethodOrder; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Optional; + import static io.restassured.RestAssured.given; import static org.hamcrest.Matchers.containsString; import static org.mockito.ArgumentMatchers.any; @@ -107,10 +111,10 @@ void testClusterAndSnapshotLifecycle() { when(containerManager.start(any(), eq("cluster-src"), any(), any())) .thenReturn(new RedshiftContainerHandle("c1", "cluster-src", "localhost", 5439)); org.mockito.Mockito.doAnswer(invocation -> { - java.nio.file.Path p = invocation.getArgument(3); - java.nio.file.Files.writeString(p, "-- dump sql table test_data;"); + Path p = invocation.getArgument(3); + Files.writeString(p, "-- dump sql table test_data;"); return null; - }).when(containerManager).takeSnapshot(any(), eq("cluster-src"), eq("admin"), any(java.nio.file.Path.class)); + }).when(containerManager).takeSnapshot(any(), eq("cluster-src"), eq("admin"), any(Path.class)); when(containerManager.start(any(), eq("cluster-restored"), any(), any())) .thenReturn(new RedshiftContainerHandle("c2", "cluster-restored", "localhost", 5440)); @@ -133,7 +137,7 @@ void testClusterAndSnapshotLifecycle() { // 1b. RebootCluster — must preserve data (no Docker volume backs this container) when(containerManager.getContainer(any(), eq("cluster-src"))) - .thenReturn(java.util.Optional.of(new RedshiftContainerHandle("c1", "cluster-src", "localhost", 5439))); + .thenReturn(Optional.of(new RedshiftContainerHandle("c1", "cluster-src", "localhost", 5439))); given() .contentType("application/x-www-form-urlencoded") .header("Authorization", AUTH_HEADER) @@ -216,8 +220,7 @@ void testClusterAndSnapshotLifecycle() { .statusCode(200) .contentType("application/xml") .body(containsString("cluster-restored")) - .body(containsString("available")) - .body(containsString("5440")); + .body(containsString("available")); // 6. DeleteClusterSnapshot given() diff --git a/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftProxyIntegrationTest.java b/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftProxyIntegrationTest.java new file mode 100644 index 0000000000..97f1677144 --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftProxyIntegrationTest.java @@ -0,0 +1,127 @@ +package io.github.hectorvent.floci.services.redshift; + +import io.github.hectorvent.floci.services.redshift.model.Cluster; +import io.quarkus.test.junit.QuarkusTest; +import jakarta.inject.Inject; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.time.Duration; +import java.util.Objects; +import org.awaitility.Awaitility; +import org.awaitility.core.ConditionTimeoutException; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@QuarkusTest +class RedshiftProxyIntegrationTest { + + @Inject + RedshiftService service; + + private String clusterId; + + @AfterEach + void cleanUp() { + if (clusterId != null) { + service.deleteCluster(clusterId); + } + } + + private static String jdbcUrl(Cluster c) { + // Use 127.0.0.1 explicitly instead of c.getEndpoint().getAddress() to avoid UnknownHostException + // in CI environments where floci.emulator.hostname is set to host.docker.internal. + return "jdbc:postgresql://127.0.0.1:" + c.getEndpoint().getPort() + "/dev"; + } + + private static Connection waitForConnection(Cluster cluster, String username, String password) throws SQLException { + try { + return Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .pollInterval(Duration.ofMillis(500)) + .ignoreExceptions() + .until(() -> DriverManager.getConnection(jdbcUrl(cluster), username, password), Objects::nonNull); + } catch (ConditionTimeoutException e) { + return DriverManager.getConnection(jdbcUrl(cluster), username, password); // throw original + } + } + + @Test + void roundTripsSqlThroughTheAdvertisedEndpoint() throws SQLException { + clusterId = "it-proxy-roundtrip"; + Cluster cluster = service.createCluster(clusterId, "dc2.large", "admin", "Secret123"); + + try (Connection conn = waitForConnection(cluster, "admin", "Secret123"); + Statement st = conn.createStatement()) { + st.execute("CREATE TABLE people (name text)"); + st.execute("INSERT INTO people VALUES ('Alice')"); + try (ResultSet rs = st.executeQuery("SELECT count(*) FROM people")) { + assertTrue(rs.next()); + assertEquals(1, rs.getInt(1)); + } + } + } + + @Test + void rejectsAWrongPasswordAtTheProxy() throws SQLException { + clusterId = "it-proxy-badpass"; + Cluster cluster = service.createCluster(clusterId, "dc2.large", "admin", "Secret123"); + + // Ensure it's ready first + try (Connection conn = waitForConnection(cluster, "admin", "Secret123")) { + assertTrue(conn.isValid(5)); + } + + assertThrows(SQLException.class, () -> + DriverManager.getConnection(jdbcUrl(cluster), "admin", "wrong-password")); + } + + @Test + void reflectsAPasswordChangeForNewConnections() throws SQLException { + clusterId = "it-proxy-rotate"; + Cluster cluster = service.createCluster(clusterId, "dc2.large", "admin", "Secret123"); + + // Ensure it's ready before testing rotation + try (Connection conn = waitForConnection(cluster, "admin", "Secret123")) { + assertTrue(conn.isValid(5)); + } + + service.modifyCluster(clusterId, null, null, "Rotated123", null, null); + + assertThrows(SQLException.class, () -> + DriverManager.getConnection(jdbcUrl(cluster), "admin", "Secret123")); + try (Connection conn = DriverManager.getConnection(jdbcUrl(cluster), "admin", "Rotated123")) { + assertTrue(conn.isValid(5)); + } + } + + @Test + void keepsTheEndpointStableAndDataIntactAcrossAReboot() throws SQLException { + clusterId = "it-proxy-reboot"; + Cluster cluster = service.createCluster(clusterId, "dc2.large", "admin", "Secret123"); + int portBefore = cluster.getEndpoint().getPort(); + + try (Connection conn = waitForConnection(cluster, "admin", "Secret123"); + Statement st = conn.createStatement()) { + st.execute("CREATE TABLE t (n int)"); + st.execute("INSERT INTO t VALUES (42)"); + } + + Cluster rebooted = service.rebootCluster(clusterId); + assertEquals(portBefore, rebooted.getEndpoint().getPort()); + + try (Connection conn = waitForConnection(rebooted, "admin", "Secret123"); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("SELECT n FROM t")) { + assertTrue(rs.next()); + assertEquals(42, rs.getInt(1)); + } + } +} diff --git a/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftServiceTest.java b/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftServiceTest.java index 23f4bca6d3..64850dc194 100644 --- a/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftServiceTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/redshift/RedshiftServiceTest.java @@ -15,6 +15,11 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -35,6 +40,8 @@ class RedshiftServiceTest { private AccountAwareStorageBackend subnetGroupBackend; private RedshiftContainerManager cm; private io.github.hectorvent.floci.core.common.RegionResolver regionResolver; + private io.github.hectorvent.floci.services.redshift.proxy.RedshiftProxyManager proxyManager; + private io.github.hectorvent.floci.core.common.docker.DockerHostResolver dockerHostResolver; private RedshiftService service; @BeforeEach @@ -47,12 +54,25 @@ void setUp() { parameterGroupBackend = mock(AccountAwareStorageBackend.class); subnetGroupBackend = mock(AccountAwareStorageBackend.class); cm = mock(RedshiftContainerManager.class); - + proxyManager = mock(io.github.hectorvent.floci.services.redshift.proxy.RedshiftProxyManager.class); + dockerHostResolver = mock(io.github.hectorvent.floci.core.common.docker.DockerHostResolver.class); + when(dockerHostResolver.resolve()).thenReturn("localhost"); + io.github.hectorvent.floci.config.EmulatorConfig config = mock(io.github.hectorvent.floci.config.EmulatorConfig.class); io.github.hectorvent.floci.config.EmulatorConfig.StorageConfig storageConfig = mock(io.github.hectorvent.floci.config.EmulatorConfig.StorageConfig.class); when(config.storage()).thenReturn(storageConfig); when(storageConfig.persistentPath()).thenReturn("target/test-data"); + io.github.hectorvent.floci.config.EmulatorConfig.ServicesConfig servicesConfig = + mock(io.github.hectorvent.floci.config.EmulatorConfig.ServicesConfig.class); + io.github.hectorvent.floci.config.EmulatorConfig.RedshiftServiceConfig redshiftConfig = + mock(io.github.hectorvent.floci.config.EmulatorConfig.RedshiftServiceConfig.class); + when(config.services()).thenReturn(servicesConfig); + when(servicesConfig.redshift()).thenReturn(redshiftConfig); + when(redshiftConfig.proxyBasePort()).thenReturn(7100); + when(redshiftConfig.proxyMaxPort()).thenReturn(7199); + when(redshiftConfig.endpointHost()).thenReturn(Optional.empty()); + when(sf.create(eq("redshift"), eq("redshift-clusters.json"), any())).thenReturn(clusterBackend); when(sf.create(eq("redshift"), eq("redshift-snapshots.json"), any())).thenReturn(snapshotBackend); when(sf.create(eq("redshift"), eq("redshift-parameter-groups.json"), any())).thenReturn(parameterGroupBackend); @@ -61,12 +81,12 @@ void setUp() { regionResolver = new io.github.hectorvent.floci.core.common.RegionResolver("us-east-1", "111111111111"); - service = new RedshiftService(sf, cm, config, regionResolver); + service = new RedshiftService(sf, cm, config, regionResolver, proxyManager, dockerHostResolver); } /** Absolute dump path as {@code createSnapshot} now stores it: under {@code /redshift-dumps/}. */ private static String dumpPath(String snapshotId) { - return java.nio.file.Paths.get("target/test-data", "redshift-dumps", "111111111111", snapshotId + ".sql") + return Paths.get("target/test-data", "redshift-dumps", "111111111111", snapshotId + ".sql") .toAbsolutePath().normalize().toString(); } @@ -138,6 +158,34 @@ void testOnStartMarksClusterUnavailableOnStartFailure() { assertEquals("unavailable", captor.getValue().getClusterStatus()); } + @Test + @SuppressWarnings("unchecked") + void onStartRestartsTheProxyForAnAdoptedCluster() { + Cluster persisted = new Cluster(); + persisted.setClusterIdentifier("c1"); + persisted.setMasterUsername("admin"); + persisted.setMasterPassword("Secret123"); + persisted.setClusterStatus("available"); + persisted.setProxyPort(7108); + + var entry = mock(AccountAwareStorageBackend.AccountEntry.class); + when(entry.value()).thenReturn(persisted); + when(entry.accountId()).thenReturn("111111111111"); + when(entry.key()).thenReturn("c1"); + when(clusterBackend.scanAllAccountEntries(any())).thenReturn(List.of(entry)); + when(cm.getContainer("111111111111", "c1")).thenReturn(Optional.empty()); + RedshiftContainerHandle handle = mock(RedshiftContainerHandle.class); + when(handle.getHost()).thenReturn("172.17.0.12"); + when(handle.getPort()).thenReturn(32830); + when(cm.adoptOrStart("111111111111", "c1", "admin", "Secret123")).thenReturn(handle); + + service.onStart(null); + + verify(proxyManager).startProxy(eq("111111111111:c1"), eq(7108), + eq("172.17.0.12"), eq(32830), eq("localhost"), + eq("admin"), eq("Secret123"), eq("dev"), any()); + } + @Test void testCreateCluster() { when(clusterBackend.get(anyString())).thenReturn(Optional.empty()); @@ -163,6 +211,29 @@ void testCreateClusterWithVpcMetadata() { assertEquals(List.of("sg-1", "sg-2"), cluster.getVpcSecurityGroupIds()); } + @Test + void createClusterStartsAProxyAndAdvertisesTheProxyEndpoint() { + when(clusterBackend.get("c1")).thenReturn(Optional.empty()); + when(clusterBackend.accountId()).thenReturn("111111111111"); + RedshiftContainerHandle handle = mock(RedshiftContainerHandle.class); + when(handle.getHost()).thenReturn("172.17.0.9"); + when(handle.getPort()).thenReturn(32800); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("Secret123"))).thenReturn(handle); + + Cluster cluster = service.createCluster("c1", "dc2.large", "admin", "Secret123"); + + // Endpoint is the proxy, not the container. + assertEquals("localhost", cluster.getEndpoint().getAddress()); + assertTrue(cluster.getEndpoint().getPort() >= 7100 && cluster.getEndpoint().getPort() <= 7199); + assertEquals("172.17.0.9", cluster.getContainerHost()); + assertEquals(32800, cluster.getContainerPort()); + assertEquals(cluster.getEndpoint().getPort(), cluster.getProxyPort()); + + verify(proxyManager).startProxy(eq("111111111111:c1"), eq(cluster.getProxyPort()), + eq("172.17.0.9"), eq(32800), eq("localhost"), + eq("admin"), eq("Secret123"), eq("dev"), any()); + } + @Test void testCreateClusterAlreadyExists() { when(clusterBackend.get("existing-cluster")).thenReturn(Optional.of(new Cluster())); @@ -171,6 +242,84 @@ void testCreateClusterAlreadyExists() { service.createCluster("existing-cluster", "dc2.large", "admin", "password123")); } + @Test + void createClusterRemovesMetadataOnFailure() { + when(clusterBackend.accountId()).thenReturn("111111111111"); + when(clusterBackend.get("c1")).thenReturn(Optional.empty()); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("password123"))) + .thenThrow(new RuntimeException("startup failed")); + + assertThrows(AwsException.class, () -> + service.createCluster("c1", "dc2.large", "admin", "password123")); + + verify(clusterBackend).delete("c1"); + verify(clusterBackend, atLeastOnce()).flush(); + } + + @Test + void createClusterKeepsMetadataWhenProxyStopFails() { + when(clusterBackend.accountId()).thenReturn("111111111111"); + when(clusterBackend.get("c1")).thenReturn(Optional.empty()); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("password123"))) + .thenThrow(new RuntimeException("startup failed")); + doThrow(new RuntimeException("proxy stop failed")) + .when(proxyManager).stopProxy("111111111111:c1"); + + assertThrows(AwsException.class, () -> + service.createCluster("c1", "dc2.large", "admin", "password123")); + + verify(clusterBackend, never()).delete("c1"); + verify(clusterBackend, atLeast(1)).put(eq("c1"), argThat(c -> "failed".equals(c.getClusterStatus()))); + verify(clusterBackend, atLeastOnce()).flush(); + } + + @Test + void restoreFromClusterSnapshotRemovesMetadataOnFailure() { + when(clusterBackend.accountId()).thenReturn("111111111111"); + Snapshot snapshot = new Snapshot(); + snapshot.setSnapshotIdentifier("snap-1"); + snapshot.setClusterIdentifier("source-c1"); + snapshot.setMasterUsername("admin"); + snapshot.setMasterPassword("password123"); + snapshot.setSqlDump(null); + + when(clusterBackend.get("c1")).thenReturn(Optional.empty()); + when(snapshotBackend.get("snap-1")).thenReturn(Optional.of(snapshot)); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("password123"))) + .thenThrow(new RuntimeException("startup failed")); + + assertThrows(AwsException.class, () -> + service.restoreFromClusterSnapshot("c1", "snap-1")); + + verify(clusterBackend).delete("c1"); + verify(clusterBackend, atLeastOnce()).flush(); + } + + @Test + void restoreFromClusterSnapshotKeepsMetadataWhenProxyStopFails() { + when(clusterBackend.accountId()).thenReturn("111111111111"); + Snapshot snapshot = new Snapshot(); + snapshot.setSnapshotIdentifier("snap-1"); + snapshot.setClusterIdentifier("source-c1"); + snapshot.setMasterUsername("admin"); + snapshot.setMasterPassword("password123"); + snapshot.setSqlDump(null); + + when(clusterBackend.get("c1")).thenReturn(Optional.empty()); + when(snapshotBackend.get("snap-1")).thenReturn(Optional.of(snapshot)); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("password123"))) + .thenThrow(new RuntimeException("startup failed")); + doThrow(new RuntimeException("proxy stop failed")) + .when(proxyManager).stopProxy("111111111111:c1"); + + assertThrows(AwsException.class, () -> + service.restoreFromClusterSnapshot("c1", "snap-1")); + + verify(clusterBackend, never()).delete("c1"); + verify(clusterBackend, atLeast(1)).put(eq("c1"), argThat(c -> "failed".equals(c.getClusterStatus()))); + verify(clusterBackend, atLeastOnce()).flush(); + } + @Test void testDescribeClusters() { Cluster c = new Cluster(); @@ -194,6 +343,41 @@ void testDeleteCluster() { verify(clusterBackend).delete("test-c"); } + @Test + void deleteClusterAbortsAndKeepsMetadataWhenTheProxyWontStop() { + Cluster c = new Cluster(); + c.setClusterIdentifier("test-c"); + c.setProxyPort(7107); + when(clusterBackend.get("test-c")).thenReturn(Optional.of(c)); + doThrow(new RuntimeException("listener close failed")) + .when(proxyManager).stopProxy("111111111111:test-c"); + + AwsException ex = assertThrows(AwsException.class, () -> service.deleteCluster("test-c")); + assertEquals("InternalFailure", ex.getErrorCode()); + + // Container and metadata are left intact so the deletion can be retried. + verify(cm, never()).stop(anyString(), anyString()); + verify(clusterBackend, never()).delete(anyString()); + } + + @Test + void deleteClusterRetrySucceedsOnceTheProxyStops() { + Cluster c = new Cluster(); + c.setClusterIdentifier("test-c"); + c.setProxyPort(7107); + when(clusterBackend.get("test-c")).thenReturn(Optional.of(c)); + doThrow(new RuntimeException("listener close failed")) + .doNothing() + .when(proxyManager).stopProxy("111111111111:test-c"); + + assertThrows(AwsException.class, () -> service.deleteCluster("test-c")); + Cluster deleted = service.deleteCluster("test-c"); + + assertEquals("deleting", deleted.getClusterStatus()); + verify(cm).stop("111111111111", "test-c"); + verify(clusterBackend).delete("test-c"); + } + @Test void testRebootClusterDumpsAndRestoresData() throws Exception { Cluster cluster = new Cluster(); @@ -205,18 +389,46 @@ void testRebootClusterDumpsAndRestoresData() throws Exception { when(cm.start(eq("111111111111"), eq("my-cluster"), eq("admin"), eq("pw"))) .thenReturn(new RedshiftContainerHandle("c-rebooted", "my-cluster", "localhost", 5555)); doAnswer(invocation -> { - java.nio.file.Path dumpFile = invocation.getArgument(3); - java.nio.file.Files.writeString(dumpFile, "-- dump"); + Path dumpFile = invocation.getArgument(3); + Files.writeString(dumpFile, "-- dump"); return null; - }).when(cm).takeSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(java.nio.file.Path.class)); + }).when(cm).takeSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(Path.class)); Cluster rebooted = service.rebootCluster("my-cluster"); assertEquals("available", rebooted.getClusterStatus()); - assertEquals(5555, rebooted.getEndpoint().getPort()); + // Endpoint now advertises the auth proxy; the restarted container is tracked separately. + assertEquals(5555, rebooted.getContainerPort()); + assertTrue(rebooted.getEndpoint().getPort() >= 7100 && rebooted.getEndpoint().getPort() <= 7199); verify(cm).stop("111111111111", "my-cluster"); verify(cm).start("111111111111", "my-cluster", "admin", "pw"); - verify(cm).restoreSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(java.nio.file.Path.class)); + verify(cm).restoreSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(Path.class)); + } + + @Test + void rebootClusterRestartsTheProxyOnTheSamePort() { + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setMasterUsername("admin"); + cluster.setMasterPassword("Secret123"); + cluster.setProxyPort(7107); + cluster.setEndpoint(new Endpoint("localhost", 7107)); + when(clusterBackend.get("c1")).thenReturn(Optional.of(cluster)); + when(clusterBackend.accountId()).thenReturn("111111111111"); + RedshiftContainerHandle handle = mock(RedshiftContainerHandle.class); + when(handle.getHost()).thenReturn("172.17.0.11"); + when(handle.getPort()).thenReturn(32820); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("Secret123"))).thenReturn(handle); + + Cluster rebooted = service.rebootCluster("c1"); + + assertEquals(7107, rebooted.getEndpoint().getPort()); + assertEquals("localhost", rebooted.getEndpoint().getAddress()); + assertEquals("172.17.0.11", rebooted.getContainerHost()); + verify(proxyManager).stopProxy("111111111111:c1"); + verify(proxyManager).startProxy(eq("111111111111:c1"), eq(7107), + eq("172.17.0.11"), eq(32820), eq("localhost"), + eq("admin"), eq("Secret123"), eq("dev"), any()); } @Test @@ -226,6 +438,94 @@ void testRebootClusterNotFound() { assertThrows(AwsException.class, () -> service.rebootCluster("missing")); } + @Test + void rebootClusterTearsDownTheProxyWhenRestoreFails() { + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setMasterUsername("admin"); + cluster.setMasterPassword("Secret123"); + cluster.setProxyPort(7107); + when(clusterBackend.get("c1")).thenReturn(Optional.of(cluster)); + RedshiftContainerHandle handle = mock(RedshiftContainerHandle.class); + when(handle.getHost()).thenReturn("172.17.0.11"); + when(handle.getPort()).thenReturn(32820); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("Secret123"))).thenReturn(handle); + doThrow(new RuntimeException("restore boom")) + .when(cm).restoreSnapshot(eq("111111111111"), eq("c1"), eq("admin"), any(Path.class)); + + assertThrows(AwsException.class, () -> service.rebootCluster("c1")); + + // The proxy started during the reboot must be stopped again on failure, and the + // replacement container must be stopped too — once before the restart, once in + // rollback — so it is not left running behind a "failed" cluster. + verify(proxyManager).startProxy(eq("111111111111:c1"), eq(7107), any(), anyInt(), + any(), any(), any(), any(), any()); + verify(proxyManager, times(2)).stopProxy("111111111111:c1"); + verify(cm, times(2)).stop("111111111111", "c1"); + } + + @Test + void rebootClusterLeavesTheOriginalContainerAloneWhenTheDumpFails() { + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setMasterUsername("admin"); + cluster.setMasterPassword("Secret123"); + cluster.setProxyPort(7107); + when(clusterBackend.get("c1")).thenReturn(Optional.of(cluster)); + doThrow(new RuntimeException("dump boom")) + .when(cm).takeSnapshot(eq("111111111111"), eq("c1"), eq("admin"), any(Path.class)); + + assertThrows(AwsException.class, () -> service.rebootCluster("c1")); + + // The dump failed before the original was torn down: rollback must not stop the + // still-running original container or touch its proxy. + verify(cm, never()).stop(anyString(), anyString()); + verify(proxyManager, never()).stopProxy(anyString()); + } + + @Test + void rebootClusterRemovesTheReplacementWhenItsStartupThrows() { + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setMasterUsername("admin"); + cluster.setMasterPassword("Secret123"); + cluster.setProxyPort(7107); + when(clusterBackend.get("c1")).thenReturn(Optional.of(cluster)); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("Secret123"))) + .thenThrow(new RuntimeException("readiness timed out")); + + assertThrows(AwsException.class, () -> service.rebootCluster("c1")); + + // start() can create the container before throwing (readiness check); the + // original is already gone, so rollback removes anything under the name — + // once for the original teardown, once for the possible orphan. + verify(cm, times(2)).stop("111111111111", "c1"); + + // The reserved port belongs to the cluster and is not released by reboot rollback; + // it is released only when the cluster is successfully deleted. + assertEquals(7107, cluster.getProxyPort()); + } + + @Test + void rebootClusterKeepsProxyPortWhenRollbackStopFails() { + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setMasterUsername("admin"); + cluster.setMasterPassword("Secret123"); + cluster.setProxyPort(7107); + when(clusterBackend.accountId()).thenReturn("111111111111"); + when(clusterBackend.get("c1")).thenReturn(Optional.of(cluster)); + when(cm.start(eq("111111111111"), eq("c1"), eq("admin"), eq("Secret123"))) + .thenThrow(new RuntimeException("readiness timed out")); + + // The first call stops the proxy during takeSnapshot; the second call happens in rollback. + doNothing().doThrow(new RuntimeException("stop failed")).when(proxyManager).stopProxy(anyString()); + + assertThrows(AwsException.class, () -> service.rebootCluster("c1")); + + assertEquals(7107, cluster.getProxyPort()); + } + @Test void testCreateSnapshot() { Cluster cluster = new Cluster(); @@ -236,7 +536,7 @@ void testCreateSnapshot() { when(clusterBackend.get("my-cluster")).thenReturn(Optional.of(cluster)); when(snapshotBackend.get("my-snapshot")).thenReturn(Optional.empty()); - doNothing().when(cm).takeSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(java.nio.file.Path.class)); + doNothing().when(cm).takeSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(Path.class)); Snapshot snapshot = service.createSnapshot("my-snapshot", "my-cluster"); assertNotNull(snapshot); @@ -252,7 +552,7 @@ void testCreateSnapshot() { assertTrue(snapshot.getSqlDump().endsWith("my-snapshot.sql")); verify(snapshotBackend).put(eq("my-snapshot"), any(Snapshot.class)); verify(snapshotBackend).flush(); - verify(cm).takeSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(java.nio.file.Path.class)); + verify(cm).takeSnapshot(eq("111111111111"), eq("my-cluster"), eq("admin"), any(Path.class)); } @Test @@ -292,7 +592,7 @@ void testCreateSnapshotRejectsTraversalOrMalformedIdentifier() { assertEquals("InvalidParameterValue", ex.getErrorCode(), id); assertEquals(400, ex.getHttpStatus(), id); } - verify(cm, never()).takeSnapshot(any(), any(), any(), any(java.nio.file.Path.class)); + verify(cm, never()).takeSnapshot(any(), any(), any(), any(Path.class)); verify(snapshotBackend, never()).put(anyString(), any(Snapshot.class)); } @@ -386,7 +686,7 @@ void testRestoreFromClusterSnapshot() { when(snapshotBackend.get("my-snapshot")).thenReturn(Optional.of(snapshot)); when(cm.start(eq("111111111111"), eq("restored-cluster"), eq("admin"), eq("password123"))) .thenReturn(new RedshiftContainerHandle("c-new", "restored-cluster", "localhost", 5432)); - doNothing().when(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(java.nio.file.Path.class)); + doNothing().when(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(Path.class)); Cluster cluster = service.restoreFromClusterSnapshot("restored-cluster", "my-snapshot", "dc2.large"); assertNotNull(cluster); @@ -399,7 +699,7 @@ void testRestoreFromClusterSnapshot() { // Restore must use the source cluster's actual password, not a hardcoded one verify(cm).start("111111111111", "restored-cluster", "admin", "password123"); - verify(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(java.nio.file.Path.class)); + verify(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(Path.class)); verify(clusterBackend, times(2)).put(eq("restored-cluster"), any(Cluster.class)); verify(clusterBackend, times(2)).flush(); } @@ -415,7 +715,7 @@ void testRestoreFromClusterSnapshotUsesStoredPasswordAfterSourceClusterDeleted() when(snapshotBackend.get("my-snapshot")).thenReturn(Optional.of(snapshot)); when(cm.start(eq("111111111111"), eq("restored-cluster"), eq("admin"), eq("original-secret"))) .thenReturn(new RedshiftContainerHandle("c-new", "restored-cluster", "localhost", 5432)); - doNothing().when(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(java.nio.file.Path.class)); + doNothing().when(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(Path.class)); Cluster cluster = service.restoreFromClusterSnapshot("restored-cluster", "my-snapshot", "dc2.large"); assertEquals("original-secret", cluster.getMasterPassword()); @@ -431,7 +731,7 @@ void testRestoreFromClusterSnapshotFallsBackToAdminWhenSourceClusterGone() { when(snapshotBackend.get("my-snapshot")).thenReturn(Optional.of(snapshot)); when(cm.start(eq("111111111111"), eq("restored-cluster"), eq("admin"), eq("admin"))) .thenReturn(new RedshiftContainerHandle("c-new", "restored-cluster", "localhost", 5432)); - doNothing().when(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(java.nio.file.Path.class)); + doNothing().when(cm).restoreSnapshot(eq("111111111111"), eq("restored-cluster"), eq("admin"), any(Path.class)); Cluster cluster = service.restoreFromClusterSnapshot("restored-cluster", "my-snapshot", "dc2.large"); assertEquals("admin", cluster.getMasterPassword()); @@ -483,6 +783,58 @@ void testRestoreFromClusterSnapshotRejectsUntrustedDumpPathBeforeProvisioning() verify(clusterBackend, never()).put(anyString(), any(Cluster.class)); } + @Test + void deleteClusterStopsTheProxyAndReleasesThePort() { + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setProxyPort(7105); + when(clusterBackend.get("c1")).thenReturn(Optional.of(cluster)); + when(clusterBackend.accountId()).thenReturn("111111111111"); + + service.deleteCluster("c1"); + + verify(proxyManager).stopProxy("111111111111:c1"); + } + + @Test + void modifyClusterPasswordUpdatesTheProxySnapshot() { + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setMasterUsername("admin"); + cluster.setMasterPassword("old"); + when(clusterBackend.get("c1")).thenReturn(Optional.of(cluster)); + when(clusterBackend.accountId()).thenReturn("111111111111"); + + service.modifyCluster("c1", null, null, "NewSecret1", null, null); + + verify(proxyManager).updateMasterPassword("111111111111:c1", "NewSecret1"); + } + + @Test + void restoreFromSnapshotStartsAProxyForTheNewCluster() { + Snapshot snap = new Snapshot(); + snap.setSnapshotIdentifier("s1"); + snap.setClusterIdentifier("src"); + snap.setMasterUsername("admin"); + snap.setMasterPassword("Secret123"); + snap.setSqlDump(null); + when(snapshotBackend.get("s1")).thenReturn(Optional.of(snap)); + when(clusterBackend.get("restored")).thenReturn(Optional.empty()); + when(clusterBackend.accountId()).thenReturn("111111111111"); + RedshiftContainerHandle handle = mock(RedshiftContainerHandle.class); + when(handle.getHost()).thenReturn("172.17.0.10"); + when(handle.getPort()).thenReturn(32810); + when(cm.start(eq("111111111111"), eq("restored"), eq("admin"), eq("Secret123"))).thenReturn(handle); + + Cluster restored = service.restoreFromClusterSnapshot("restored", "s1"); + + assertEquals("localhost", restored.getEndpoint().getAddress()); + assertEquals(restored.getEndpoint().getPort(), restored.getProxyPort()); + verify(proxyManager).startProxy(eq("111111111111:restored"), anyInt(), + eq("172.17.0.10"), eq(32810), eq("localhost"), + eq("admin"), eq("Secret123"), eq("dev"), any()); + } + @Test void testCreateClusterParameterGroup() { when(parameterGroupBackend.get("my-pg")).thenReturn(Optional.empty()); @@ -580,7 +932,7 @@ void testModifyClusterParameterGroupNotFound() { @Test void testDescribeClusterParametersReturnsStoredValues() { ClusterParameterGroup group = new ClusterParameterGroup("my-pg", "redshift-1.0", "custom pg"); - group.setParameters(new java.util.ArrayList<>(List.of(new Parameter("statement_timeout", "5000")))); + group.setParameters(new ArrayList<>(List.of(new Parameter("statement_timeout", "5000")))); when(parameterGroupBackend.get("my-pg")).thenReturn(Optional.of(group)); List params = service.describeClusterParameters("my-pg"); @@ -614,7 +966,7 @@ void testCreateAndListTagsForCluster() { when(clusterBackend.get("my-cluster")).thenReturn(Optional.of(cluster)); service.createTags("arn:aws:redshift:us-east-1:111111111111:cluster:my-cluster", - java.util.Map.of("env", "test")); + Map.of("env", "test")); assertEquals("test", cluster.getTags().get("env")); verify(clusterBackend).put(eq("my-cluster"), any(Cluster.class)); @@ -624,31 +976,31 @@ void testCreateAndListTagsForCluster() { void testDeleteTagsForCluster() { Cluster cluster = new Cluster(); cluster.setClusterIdentifier("my-cluster"); - cluster.setTags(new java.util.LinkedHashMap<>(java.util.Map.of("env", "test", "team", "data"))); + cluster.setTags(new LinkedHashMap<>(Map.of("env", "test", "team", "data"))); when(clusterBackend.get("my-cluster")).thenReturn(Optional.of(cluster)); - service.deleteTags("arn:aws:redshift:us-east-1:111111111111:cluster:my-cluster", java.util.List.of("env")); + service.deleteTags("arn:aws:redshift:us-east-1:111111111111:cluster:my-cluster", List.of("env")); - assertEquals(java.util.Map.of("team", "data"), cluster.getTags()); + assertEquals(Map.of("team", "data"), cluster.getTags()); } @Test void testCreateTagsRejectsNonArnResourceName() { assertThrows(AwsException.class, () -> - service.createTags("my-cluster", java.util.Map.of("env", "test"))); + service.createTags("my-cluster", Map.of("env", "test"))); } @Test void testCreateTagsRejectsUnknownResourceType() { assertThrows(AwsException.class, () -> - service.createTags("arn:aws:redshift:us-east-1:111111111111:reservednode:foo", java.util.Map.of("env", "test"))); + service.createTags("arn:aws:redshift:us-east-1:111111111111:reservednode:foo", Map.of("env", "test"))); } @Test void testDescribeTagsForSpecificResource() { Cluster cluster = new Cluster(); cluster.setClusterIdentifier("my-cluster"); - cluster.setTags(new java.util.LinkedHashMap<>(java.util.Map.of("env", "test"))); + cluster.setTags(new LinkedHashMap<>(Map.of("env", "test"))); when(clusterBackend.get("my-cluster")).thenReturn(Optional.of(cluster)); List tagged = @@ -664,14 +1016,14 @@ void testDescribeTagsForSpecificResource() { void testDescribeTagsScansAllResourcesOfType() { Cluster a = new Cluster(); a.setClusterIdentifier("cluster-a"); - a.setTags(new java.util.LinkedHashMap<>(java.util.Map.of("env", "prod"))); + a.setTags(new LinkedHashMap<>(Map.of("env", "prod"))); Cluster b = new Cluster(); b.setClusterIdentifier("cluster-b"); - b.setTags(new java.util.LinkedHashMap<>()); - when(clusterBackend.scan(any())).thenReturn(java.util.List.of(a, b)); - when(snapshotBackend.scan(any())).thenReturn(java.util.List.of()); - when(parameterGroupBackend.scan(any())).thenReturn(java.util.List.of()); - when(subnetGroupBackend.scan(any())).thenReturn(java.util.List.of()); + b.setTags(new LinkedHashMap<>()); + when(clusterBackend.scan(any())).thenReturn(List.of(a, b)); + when(snapshotBackend.scan(any())).thenReturn(List.of()); + when(parameterGroupBackend.scan(any())).thenReturn(List.of()); + when(subnetGroupBackend.scan(any())).thenReturn(List.of()); List tagged = service.describeTags(null, "cluster", null); diff --git a/src/test/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManagerTest.java b/src/test/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManagerTest.java index e6da56d53f..209b2bb62a 100644 --- a/src/test/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManagerTest.java +++ b/src/test/java/io/github/hectorvent/floci/services/redshift/container/RedshiftContainerManagerTest.java @@ -23,10 +23,15 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; import java.io.Closeable; import java.io.InputStream; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Map; +import com.github.dockerjava.api.command.CopyArchiveFromContainerCmd; import java.util.Optional; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -65,6 +70,28 @@ void setUp() { when(config.services().redshift().imageVersion()).thenReturn("postgres:16-alpine"); when(config.services().redshift().dockerNetwork()).thenReturn(Optional.empty()); + // Default mock for execCreateCmd/execStartCmd/inspectExecCmd to make waitForReady succeed instantly + ExecCreateCmd defaultCreateCmd = mock(ExecCreateCmd.class, org.mockito.Mockito.RETURNS_SELF); + ExecCreateCmdResponse defaultCreateResponse = mock(ExecCreateCmdResponse.class); + when(defaultCreateResponse.getId()).thenReturn("exec-default"); + when(defaultCreateCmd.exec()).thenReturn(defaultCreateResponse); + when(dockerClient.execCreateCmd(anyString())).thenReturn(defaultCreateCmd); + + ExecStartCmd defaultStartCmd = mock(ExecStartCmd.class); + when(defaultStartCmd.exec(any())).thenAnswer(invocation -> { + @SuppressWarnings("unchecked") + ResultCallback.Adapter adapter = invocation.getArgument(0); + adapter.onComplete(); + return adapter; + }); + when(dockerClient.execStartCmd(anyString())).thenReturn(defaultStartCmd); + + InspectExecCmd defaultInspectCmd = mock(InspectExecCmd.class); + InspectExecResponse defaultInspectResponse = mock(InspectExecResponse.class); + when(defaultInspectResponse.getExitCodeLong()).thenReturn(0L); + when(defaultInspectCmd.exec()).thenReturn(defaultInspectResponse); + when(dockerClient.inspectExecCmd(anyString())).thenReturn(defaultInspectCmd); + manager = new RedshiftContainerManager( containerBuilder, lifecycleManager, @@ -79,7 +106,7 @@ void setUp() { @Test void testTakeSnapshotContainerNotFound() { AwsException ex = assertThrows(AwsException.class, () -> - manager.takeSnapshot(ACCOUNT_ID, "non-existent-cluster", "admin", "dev", java.nio.file.Path.of("dummy.sql"))); + manager.takeSnapshot(ACCOUNT_ID, "non-existent-cluster", "admin", "dev", Path.of("dummy.sql"))); assertEquals("ClusterNotFound", ex.getErrorCode()); assertEquals(404, ex.getHttpStatus()); } @@ -87,7 +114,7 @@ void testTakeSnapshotContainerNotFound() { @Test void testRestoreSnapshotContainerNotFound() { AwsException ex = assertThrows(AwsException.class, () -> - manager.restoreSnapshot(ACCOUNT_ID, "non-existent-cluster", "admin", "dev", java.nio.file.Path.of("dummy.sql"))); + manager.restoreSnapshot(ACCOUNT_ID, "non-existent-cluster", "admin", "dev", Path.of("dummy.sql"))); assertEquals("ClusterNotFound", ex.getErrorCode()); assertEquals(404, ex.getHttpStatus()); } @@ -95,7 +122,7 @@ void testRestoreSnapshotContainerNotFound() { @Test void testCreateSnapshotNullCluster() { AwsException ex = assertThrows(AwsException.class, () -> - manager.createSnapshot(ACCOUNT_ID, null, java.nio.file.Path.of("dummy.sql"))); + manager.createSnapshot(ACCOUNT_ID, null, Path.of("dummy.sql"))); assertEquals("InvalidParameterValue", ex.getErrorCode()); assertEquals(400, ex.getHttpStatus()); } @@ -103,7 +130,7 @@ void testCreateSnapshotNullCluster() { @Test void testRestoreSnapshotNullCluster() { AwsException ex = assertThrows(AwsException.class, () -> - manager.restoreSnapshot(ACCOUNT_ID, (Cluster) null, java.nio.file.Path.of("dummy.sql"))); + manager.restoreSnapshot(ACCOUNT_ID, (Cluster) null, Path.of("dummy.sql"))); assertEquals("InvalidParameterValue", ex.getErrorCode()); assertEquals(400, ex.getHttpStatus()); } @@ -143,9 +170,9 @@ void testTakeSnapshotSuccess() throws Exception { when(dockerClient.inspectExecCmd("exec-1")).thenReturn(inspectCmd); // mock copyArchiveFromContainerCmd - com.github.dockerjava.api.command.CopyArchiveFromContainerCmd copyCmd = mock(com.github.dockerjava.api.command.CopyArchiveFromContainerCmd.class, org.mockito.Mockito.RETURNS_SELF); + CopyArchiveFromContainerCmd copyCmd = mock(CopyArchiveFromContainerCmd.class, org.mockito.Mockito.RETURNS_SELF); byte[] tarBytes = "-- PostgreSQL dump\nCREATE TABLE foo (id int);\n".getBytes(StandardCharsets.UTF_8); - java.io.ByteArrayOutputStream bos = new java.io.ByteArrayOutputStream(); + ByteArrayOutputStream bos = new ByteArrayOutputStream(); try (org.apache.commons.compress.archivers.tar.TarArchiveOutputStream tar = new org.apache.commons.compress.archivers.tar.TarArchiveOutputStream(bos)) { org.apache.commons.compress.archivers.tar.TarArchiveEntry entry = new org.apache.commons.compress.archivers.tar.TarArchiveEntry("dump.sql"); entry.setSize(tarBytes.length); @@ -153,17 +180,17 @@ void testTakeSnapshotSuccess() throws Exception { tar.write(tarBytes); tar.closeArchiveEntry(); } - when(copyCmd.exec()).thenReturn(new java.io.ByteArrayInputStream(bos.toByteArray())); + when(copyCmd.exec()).thenReturn(new ByteArrayInputStream(bos.toByteArray())); when(dockerClient.copyArchiveFromContainerCmd("cont-123", "/tmp/dump.sql")).thenReturn(copyCmd); - java.nio.file.Path tempFile = java.nio.file.Files.createTempFile("test-take-snapshot", ".sql"); + Path tempFile = Files.createTempFile("test-take-snapshot", ".sql"); try { manager.takeSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",tempFile); - String dump = java.nio.file.Files.readString(tempFile); + String dump = Files.readString(tempFile); assertTrue(dump.contains("PostgreSQL dump")); assertTrue(dump.contains("CREATE TABLE foo")); } finally { - java.nio.file.Files.deleteIfExists(tempFile); + Files.deleteIfExists(tempFile); } } @@ -200,7 +227,7 @@ void testTakeSnapshotFailureExitCode() throws Exception { when(dockerClient.inspectExecCmd("exec-fail")).thenReturn(inspectCmd); AwsException ex = assertThrows(AwsException.class, () -> - manager.takeSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",java.nio.file.Path.of("dummy.sql"))); + manager.takeSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",Path.of("dummy.sql"))); assertEquals("InternalFailure", ex.getErrorCode()); assertEquals(500, ex.getHttpStatus()); } @@ -215,8 +242,8 @@ void testRestoreSnapshotEmptyDump() { manager.start(ACCOUNT_ID, "test-cluster", "admin", "pass"); // Should return cleanly without touching dockerClient - manager.restoreSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",java.nio.file.Path.of("non-existent-dump.sql")); - manager.restoreSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",(java.nio.file.Path) null); + manager.restoreSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",Path.of("non-existent-dump.sql")); + manager.restoreSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",(Path) null); } @Test @@ -254,13 +281,13 @@ void testRestoreSnapshotSuccess() throws Exception { when(inspectCmd.exec()).thenReturn(inspectResponse); when(dockerClient.inspectExecCmd("exec-restore")).thenReturn(inspectCmd); - java.nio.file.Path tempFile = java.nio.file.Files.createTempFile("test-restore-snapshot", ".sql"); + Path tempFile = Files.createTempFile("test-restore-snapshot", ".sql"); try { manager.restoreSnapshot(ACCOUNT_ID, "test-cluster", "admin", "dev",tempFile); verify(dockerClient).copyArchiveToContainerCmd("cont-123"); - verify(dockerClient).execCreateCmd("cont-123"); + verify(dockerClient, org.mockito.Mockito.times(2)).execCreateCmd("cont-123"); } finally { - java.nio.file.Files.deleteIfExists(tempFile); + Files.deleteIfExists(tempFile); } } @@ -363,7 +390,7 @@ void testAlterUserPasswordSuccess() throws Exception { manager.alterUserPassword(ACCOUNT_ID, "test-cluster", "admin", "NewSecret1"); - verify(dockerClient).execCreateCmd("cont-123"); + verify(dockerClient, org.mockito.Mockito.times(2)).execCreateCmd("cont-123"); } @Test diff --git a/src/test/java/io/github/hectorvent/floci/services/redshift/model/ClusterTest.java b/src/test/java/io/github/hectorvent/floci/services/redshift/model/ClusterTest.java new file mode 100644 index 0000000000..dfc7028f6e --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/redshift/model/ClusterTest.java @@ -0,0 +1,45 @@ +package io.github.hectorvent.floci.services.redshift.model; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +class ClusterTest { + + @Test + void proxyAndContainerFieldsDefaultToEmpty() { + Cluster cluster = new Cluster(); + assertNull(cluster.getContainerHost()); + assertEquals(0, cluster.getContainerPort()); + assertEquals(0, cluster.getProxyPort()); + } + + @Test + void proxyAndContainerFieldsRoundTripThroughJackson() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + Cluster cluster = new Cluster(); + cluster.setClusterIdentifier("c1"); + cluster.setContainerHost("172.17.0.4"); + cluster.setContainerPort(32771); + cluster.setProxyPort(7100); + + Cluster restored = mapper.readValue(mapper.writeValueAsString(cluster), Cluster.class); + + assertEquals("172.17.0.4", restored.getContainerHost()); + assertEquals(32771, restored.getContainerPort()); + assertEquals(7100, restored.getProxyPort()); + } + + @Test + void jsonWithoutNewFieldsDeserializesWithDefaults() throws Exception { + ObjectMapper mapper = new ObjectMapper(); + Cluster restored = mapper.readValue( + "{\"clusterIdentifier\":\"legacy\",\"clusterStatus\":\"available\"}", Cluster.class); + + assertNull(restored.getContainerHost()); + assertEquals(0, restored.getContainerPort()); + assertEquals(0, restored.getProxyPort()); + } +} diff --git a/src/test/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftAuthProxyTest.java b/src/test/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftAuthProxyTest.java new file mode 100644 index 0000000000..804e5d3a08 --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftAuthProxyTest.java @@ -0,0 +1,205 @@ +package io.github.hectorvent.floci.services.redshift.proxy; + +import io.github.hectorvent.floci.config.EmulatorConfig; +import io.github.hectorvent.floci.services.acm.CertificateGenerator; +import io.github.hectorvent.floci.services.rds.proxy.RdsProxyTlsCertificates; +import io.github.hectorvent.floci.services.rds.proxy.RdsSigV4Validator; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.lang.reflect.Field; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class RedshiftAuthProxyTest { + + @TempDir + Path tempDir; + + private RedshiftAuthProxy proxy; + private ServerSocket fakeBackend; + + @AfterEach + void tearDown() throws IOException { + if (proxy != null) { + proxy.stop(); + } + if (fakeBackend != null && !fakeBackend.isClosed()) { + fakeBackend.close(); + } + } + + @Test + void bridgesClientBytesToTheBackendAfterASuccessfulPasswordAuth() throws Exception { + // The fake backend accepts one connection and records the first bytes it receives + // from the proxy (the forwarded PostgreSQL startup packet). + fakeBackend = new ServerSocket(0); + AtomicReference seenByBackend = new AtomicReference<>(); + CountDownLatch backendDone = new CountDownLatch(1); + Thread.ofVirtual().start(() -> { + try (Socket s = fakeBackend.accept()) { + InputStream in = s.getInputStream(); + byte[] buf = new byte[256]; + int n = in.read(buf); + byte[] out = new byte[Math.max(n, 0)]; + System.arraycopy(buf, 0, out, 0, out.length); + seenByBackend.set(out); + backendDone.countDown(); + } catch (IOException ignored) { + backendDone.countDown(); + } + }); + + int proxyPort = freePort(); + proxy = new RedshiftAuthProxy("111111111111:c1", "localhost", fakeBackend.getLocalPort(), + "admin", "Secret123", "dev", + mock(RdsSigV4Validator.class), realTls(), (user, pw) -> true); + proxy.start(proxyPort); + + try (Socket client = new Socket("localhost", proxyPort)) { + // Minimal PostgreSQL v3 startup: length(4) + protocol(4) + "user\0admin\0\0" + OutputStream out = client.getOutputStream(); + byte[] params = "user\0admin\0\0".getBytes(StandardCharsets.UTF_8); + int len = 8 + params.length; + out.write(new byte[]{(byte) (len >>> 24), (byte) (len >>> 16), (byte) (len >>> 8), (byte) len}); + out.write(new byte[]{0, 3, 0, 0}); // protocol 196608 + out.write(params); + out.flush(); + + // Read the AuthenticationCleartextPassword request ('R') and reply with a password message. + InputStream in = client.getInputStream(); + assertEquals('R', in.read()); + in.readNBytes(7); // remaining length(4) + auth code(3 of the 4-byte int) + + OutputStream pw = client.getOutputStream(); + byte[] pwBytes = "Secret123\0".getBytes(StandardCharsets.UTF_8); + pw.write('p'); + int pwLen = 4 + pwBytes.length; + pw.write(new byte[]{(byte) (pwLen >>> 24), (byte) (pwLen >>> 16), (byte) (pwLen >>> 8), (byte) pwLen}); + pw.write(pwBytes); + pw.flush(); + + assertTrue(backendDone.await(5, TimeUnit.SECONDS), "backend never saw the forwarded startup packet"); + byte[] forwarded = seenByBackend.get(); + assertTrue(forwarded != null && forwarded.length > 0, "proxy forwarded no bytes to the backend"); + } + } + + @Test + void closesTheBackendConnectionWhenTheClientDropsMidHandshake() throws Exception { + fakeBackend = new ServerSocket(0); + CountDownLatch backendClosed = new CountDownLatch(1); + Thread.ofVirtual().start(() -> { + try (Socket s = fakeBackend.accept()) { + // The proxy opens this before reading the client's startup packet; if the + // client vanishes, the proxy must close this too — surfaced here as EOF. + InputStream in = s.getInputStream(); + while (in.read() != -1) { + // drain until the proxy closes its end + } + backendClosed.countDown(); + } catch (IOException e) { + backendClosed.countDown(); + } + }); + + int proxyPort = freePort(); + proxy = new RedshiftAuthProxy("111111111111:c1", "localhost", fakeBackend.getLocalPort(), + "admin", "Secret123", "dev", + mock(RdsSigV4Validator.class), realTls(), (user, pw) -> true); + proxy.start(proxyPort); + + // Connect, then drop without ever sending a startup packet. + new Socket("localhost", proxyPort).close(); + + assertTrue(backendClosed.await(5, TimeUnit.SECONDS), + "proxy leaked the backend connection after the client dropped"); + } + + @Test + void startRetriesTheBindWhileThePortIsMomentarilyStillInUse() throws Exception { + fakeBackend = new ServerSocket(0); + int proxyPort = freePort(); + + // Hold the port, mimicking a just-closed predecessor the kernel has not released + // yet; free it shortly after so the retrying bind can finally take it. + ServerSocket squatter = new ServerSocket(); + squatter.setReuseAddress(true); + squatter.bind(new java.net.InetSocketAddress(proxyPort)); + Thread.ofVirtual().start(() -> { + try { + Thread.sleep(200); + squatter.close(); + } catch (Exception ignored) { + } + }); + + proxy = new RedshiftAuthProxy("111111111111:c1", "localhost", fakeBackend.getLocalPort(), + "admin", "Secret123", "dev", + mock(RdsSigV4Validator.class), realTls(), (user, pw) -> true); + proxy.start(proxyPort); // must not throw despite the port being busy at first + + assertTrue(portAccepts(proxyPort), "proxy never bound the port after the squatter released it"); + } + + @Test + void updateMasterPasswordSwapsTheSnapshotUsedForNewConnections() throws Exception { + fakeBackend = new ServerSocket(0); + int proxyPort = freePort(); + proxy = new RedshiftAuthProxy("111111111111:c1", "localhost", fakeBackend.getLocalPort(), + "admin", "old", "dev", + mock(RdsSigV4Validator.class), realTls(), (user, pw) -> true); + proxy.start(proxyPort); + + proxy.updateMasterPassword("rotated"); + + Field f = RedshiftAuthProxy.class.getDeclaredField("masterPassword"); + f.setAccessible(true); + assertEquals("rotated", f.get(proxy)); + } + + private RdsProxyTlsCertificates realTls() { + // The real bean generates a self-signed cert on demand; no Docker or network needed. + EmulatorConfig.StorageConfig storage = mock(EmulatorConfig.StorageConfig.class); + when(storage.persistentPath()).thenReturn(tempDir.toString()); + EmulatorConfig config = mock(EmulatorConfig.class); + when(config.storage()).thenReturn(storage); + return new RdsProxyTlsCertificates(config, new CertificateGenerator()); + } + + private static int freePort() throws IOException { + try (ServerSocket s = new ServerSocket(0)) { + return s.getLocalPort(); + } + } + + private static boolean portAccepts(int port) { + for (int attempt = 0; attempt < 50; attempt++) { + try (Socket ignored = new Socket("localhost", port)) { + return true; + } catch (IOException e) { + try { + Thread.sleep(20); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + return false; + } + } + } + return false; + } +} diff --git a/src/test/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftProxyManagerTest.java b/src/test/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftProxyManagerTest.java new file mode 100644 index 0000000000..0c07c91cb7 --- /dev/null +++ b/src/test/java/io/github/hectorvent/floci/services/redshift/proxy/RedshiftProxyManagerTest.java @@ -0,0 +1,287 @@ +package io.github.hectorvent.floci.services.redshift.proxy; + +import io.github.hectorvent.floci.services.rds.proxy.RdsProxyTlsCertificates; +import io.github.hectorvent.floci.services.rds.proxy.RdsSigV4Validator; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.util.concurrent.ConcurrentHashMap; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +class RedshiftProxyManagerTest { + + private RedshiftProxyManager newManager() { + return new RedshiftProxyManager( + mock(RdsSigV4Validator.class), mock(RdsProxyTlsCertificates.class)); + } + + private static void start(RedshiftProxyManager manager, String key, int proxyPort) { + manager.startProxy(key, proxyPort, "localhost", 1, "localhost", + "admin", "secret", "dev", (user, password) -> true); + } + + @Test + void startRegistersTheKeyAndBindsThePort() throws Exception { + RedshiftProxyManager manager = newManager(); + int port = availablePort(); + try { + start(manager, "111111111111:c1", port); + assertTrue(registry(manager).containsKey("111111111111:c1")); + assertPortUnavailable(port); + } finally { + manager.stopAll(); + } + } + + @Test + void stopProxyRemovesTheKeyAndReleasesThePort() throws Exception { + RedshiftProxyManager manager = newManager(); + int port = availablePort(); + start(manager, "k", port); + + manager.stopProxy("k"); + + assertFalse(registry(manager).containsKey("k")); + assertPortAvailable(port); + } + + @Test + void startingAnExistingKeyStopsTheOldProxyAndInstallsTheNewOne() throws IOException { + RedshiftProxyManager manager = newManager(); + int firstPort = availablePort(); + try { + start(manager, "k", firstPort); + int replacementPort = availablePort(); + + start(manager, "k", replacementPort); + + assertPortAvailable(firstPort); + assertPortUnavailable(replacementPort); + } finally { + manager.stopAll(); + } + } + + + @Test + void updateMasterPasswordForUnknownKeyIsANoOp() { + RedshiftProxyManager manager = newManager(); + assertDoesNotThrow(() -> manager.updateMasterPassword("missing", "rotated")); + } + + @Test + void updateMasterPasswordSwapsTheRunningSnapshot() throws Exception { + RedshiftProxyManager manager = newManager(); + int port = availablePort(); + try { + start(manager, "k", port); + manager.updateMasterPassword("k", "rotated"); + assertEquals("rotated", masterPassword(registry(manager).get("k"))); + } finally { + manager.stopAll(); + } + } + + @Test + void stopAllReleasesEveryListenerAndIsIdempotent() throws IOException { + RedshiftProxyManager manager = newManager(); + int a = availablePort(); + start(manager, "a", a); + int b = availablePort(); + start(manager, "b", b); + + manager.stopAll(); + + assertPortAvailable(a); + assertPortAvailable(b); + assertDoesNotThrow(manager::stopAll); + } + + @Test + void failedStartOnABusyPortLeavesNoRegistryEntry() throws Exception { + RedshiftProxyManager manager = newManager(); + try (ServerSocket occupied = new ServerSocket(0)) { + assertThrows(RuntimeException.class, + () -> start(manager, "k", occupied.getLocalPort())); + assertFalse(registry(manager).containsKey("k")); + } finally { + manager.stopAll(); + } + } + + @Test + void stopProxyRetainsTheProxyWhenItsListenerCloseFails() throws Exception { + RedshiftProxyManager manager = newManager(); + RedshiftAuthProxy badProxy = mock(RedshiftAuthProxy.class); + doThrow(new RuntimeException("close failed")).when(badProxy).stop(); + registry(manager).put("k", badProxy); + + assertThrows(RuntimeException.class, () -> manager.stopProxy("k")); + + // Proxy stays registered so a later cleanup attempt can still reach it and retry. + assertSame(badProxy, registry(manager).get("k")); + } + + @Test + void stopProxyRetryClosesTheSameProxyAndThenDeregistersIt() throws Exception { + RedshiftProxyManager manager = newManager(); + RedshiftAuthProxy proxy = mock(RedshiftAuthProxy.class); + doThrow(new RuntimeException("close failed")).doNothing().when(proxy).stop(); + registry(manager).put("k", proxy); + + assertThrows(RuntimeException.class, () -> manager.stopProxy("k")); + assertDoesNotThrow(() -> manager.stopProxy("k")); + + assertFalse(registry(manager).containsKey("k")); + verify(proxy, times(2)).stop(); + } + + @Test + void stopProxyRefusesWhileAFailedStartupListenerStillCannotBeClosed() throws Exception { + RedshiftProxyManager manager = newManager(); + RedshiftAuthProxy stuck = mock(RedshiftAuthProxy.class); + doThrow(new RuntimeException("close failed")).when(stuck).stop(); + unclosable(manager).put("k", stuck); + + assertThrows(RuntimeException.class, () -> manager.stopProxy("k")); + + // Reference retained so the next attempt can retry the close. + assertSame(stuck, unclosable(manager).get("k")); + } + + @Test + void stopProxyKeepsRetryingTheSameUnclosableListenerOnEveryCall() throws Exception { + RedshiftProxyManager manager = newManager(); + RedshiftAuthProxy stuck = mock(RedshiftAuthProxy.class); + doThrow(new RuntimeException("close failed")).when(stuck).stop(); + unclosable(manager).put("k", stuck); + + assertThrows(RuntimeException.class, () -> manager.stopProxy("k")); + assertThrows(RuntimeException.class, () -> manager.stopProxy("k")); + + verify(stuck, times(2)).stop(); + assertTrue(unclosable(manager).containsKey("k")); + } + + @Test + void stopProxyRecoversTheUnclosableEntryOnceTheListenerFinallyCloses() throws Exception { + RedshiftProxyManager manager = newManager(); + RedshiftAuthProxy recovering = mock(RedshiftAuthProxy.class); + doThrow(new RuntimeException("close failed")).doNothing().when(recovering).stop(); + unclosable(manager).put("k", recovering); + + assertThrows(RuntimeException.class, () -> manager.stopProxy("k")); + assertDoesNotThrow(() -> manager.stopProxy("k")); + + assertFalse(unclosable(manager).containsKey("k")); + } + + @Test + void startProxyLeavesAnUnclosableEntryInPlace() throws Exception { + RedshiftProxyManager manager = newManager(); + RedshiftAuthProxy stuck = mock(RedshiftAuthProxy.class); + doThrow(new RuntimeException("close failed")).when(stuck).stop(); + unclosable(manager).put("k", stuck); + int port = availablePort(); + try { + // A fresh start for the same key must not silently drop the leaked listener. + start(manager, "k", port); + assertSame(stuck, unclosable(manager).get("k")); + } finally { + manager.stopAll(); + } + } + + @Test + void stopAllAlsoDrainsUnclosableProxies() throws Exception { + RedshiftProxyManager manager = newManager(); + RedshiftAuthProxy recovered = mock(RedshiftAuthProxy.class); + RedshiftAuthProxy stillStuck = mock(RedshiftAuthProxy.class); + doThrow(new RuntimeException("close failed")).when(stillStuck).stop(); + unclosable(manager).put("a", recovered); + unclosable(manager).put("b", stillStuck); + + manager.stopAll(); + + assertFalse(unclosable(manager).containsKey("a")); + assertTrue(unclosable(manager).containsKey("b")); + } + + // --- reflection + port helpers copied from RdsProxyManagerTest --- + + @SuppressWarnings("unchecked") + private static ConcurrentHashMap registry(RedshiftProxyManager manager) + throws Exception { + Field field = RedshiftProxyManager.class.getDeclaredField("proxies"); + field.setAccessible(true); + return (ConcurrentHashMap) field.get(manager); + } + + @SuppressWarnings("unchecked") + private static ConcurrentHashMap unclosable(RedshiftProxyManager manager) + throws Exception { + Field field = RedshiftProxyManager.class.getDeclaredField("unclosableProxies"); + field.setAccessible(true); + return (ConcurrentHashMap) field.get(manager); + } + + private static String masterPassword(RedshiftAuthProxy proxy) throws Exception { + Field field = RedshiftAuthProxy.class.getDeclaredField("masterPassword"); + field.setAccessible(true); + return (String) field.get(proxy); + } + + private static int availablePort() throws IOException { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } + + private static void assertPortAvailable(int port) { + IOException last = null; + for (int attempt = 0; attempt < 50; attempt++) { + try (ServerSocket ignored = reusableSocket(port)) { + return; + } catch (IOException e) { + last = e; + } + try { + Thread.sleep(10); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + fail("Interrupted waiting for port " + port, e); + } + } + fail("Proxy port " + port + " was not released", last); + } + + private static void assertPortUnavailable(int port) { + assertThrows(IOException.class, () -> { + try (ServerSocket ignored = reusableSocket(port)) { + // active proxy owns this listener + } + }); + } + + private static ServerSocket reusableSocket(int port) throws IOException { + ServerSocket socket = new ServerSocket(); + socket.setReuseAddress(true); + socket.bind(new InetSocketAddress(port)); + return socket; + } +} diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml index efffba7e3e..0954f2fcaf 100644 --- a/src/test/resources/application.yml +++ b/src/test/resources/application.yml @@ -167,6 +167,8 @@ floci: enabled: true default-port: 5439 image-version: postgres:15-alpine + proxy-base-port: 7100 + proxy-max-port: 7199 rds: enabled: true mock: false