diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java index 4f34d41c9..b597333f0 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/VotingPluginMain.java @@ -152,6 +152,11 @@ public class VotingPluginMain extends AdvancedCorePlugin { @Getter public static VotingPluginMain plugin; + /** Exact backend plugin file name Bukkit uses when consuming its update folder. */ + public File getLoadedPluginJarFile() { + return getFile(); + } + @Getter @Setter private ArrayList adminVoteCommand; diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/control/BackendControlConnector.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/control/BackendControlConnector.java index 91a8a5915..ac5a1bc20 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/control/BackendControlConnector.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/control/BackendControlConnector.java @@ -34,6 +34,7 @@ import com.bencodez.votingplugin.VotingPluginMain; import com.bencodez.votingplugin.control.BackendControlResultStore.Route; import com.bencodez.votingplugin.control.BackendControlResultStore.StoredResult; +import com.bencodez.votingplugin.proxy.control.HostedControlManager; import com.bencodez.votingplugin.proxy.control.HostedControlManager.HostConfiguration; import com.bencodez.votingplugin.util.BoundedHttpBodyHandler; import com.bencodez.votingplugin.util.ControlCredentialFile; @@ -53,6 +54,7 @@ public final class BackendControlConnector implements AutoCloseable { private static final Set CAPABILITIES = Set.of("config.files.v1", "config.file-comments.v1", "config.quick-setup.v1", "config.quick-setup.v2", "config.vote-sites-sync.v1", "config.proxy-method.v1", "config.reward-files.v1", "data.inspect.v1"); + private static final Set DEPLOYMENT_TASK_FIELDS = Set.of("deploymentId", "artifactId", "sha256", "size", "attemptId"); private final VotingPluginMain plugin; private final Path dataDirectory; @@ -65,6 +67,8 @@ public final class BackendControlConnector implements AutoCloseable { private final HttpClient http; private final BackendConfigurationService configurations; private final ControlInspectionService inspections; + private final PluginDeploymentService deployments; + private final boolean directLocalDeploymentEndpoint; private final UUID sessionId = UUID.randomUUID(); private final Map completed = new LinkedHashMap<>(); private final boolean recovering; @@ -80,12 +84,14 @@ public final class BackendControlConnector implements AutoCloseable { private volatile boolean voteSitesSyncAccepted; private volatile boolean rewardFilesAccepted; private volatile boolean inspectionsAccepted; + private volatile boolean deploymentsAccepted; private volatile int inspectionFailures; private volatile long inspectionRetryAtNanos; private volatile int failures; private volatile ScheduledFuture scheduled; private volatile ScheduledFuture operationPolling; private volatile ScheduledFuture inspectionPolling; + private volatile ScheduledFuture deploymentPolling; private volatile Future activeReload; private volatile CompletableFuture activeOperation; @@ -128,6 +134,22 @@ private BackendControlConnector(VotingPluginMain plugin, Path dataDirectory, Set } }); inspections = new ControlInspectionService(plugin); + directLocalDeploymentEndpoint = HostedControlManager.isDirectLocalEndpoint( + settings.endpoint().toString(), hostedConfiguration); + PluginDeploymentService prepared = null; + boolean deploymentEndpointAllowed = PluginDeploymentService.credentialEndpointAllowed( + settings.endpoint(), directLocalDeploymentEndpoint); + if (!recovering && deploymentEndpointAllowed) { + try { + prepared = PluginDeploymentService.backend(plugin.getServer().getUpdateFolderFile().toPath(), + plugin.getLoadedPluginJarFile().toPath()); + } catch (Exception failure) { + plugin.getLogger().warning("[Control] Plugin deployment staging is unavailable; capability not advertised"); + } + } else if (!recovering && !deploymentEndpointAllowed) { + plugin.getLogger().warning("[Control] Plugin deployment staging requires HTTPS unless Control is hosted directly on this node"); + } + deployments = prepared; } private void reloadConfiguration(String fileName, String expectedContent) throws Exception { @@ -268,6 +290,76 @@ public void start() { OPERATION_POLL_MILLIS, OPERATION_POLL_MILLIS, TimeUnit.MILLISECONDS); inspectionPolling = inspectionExecutor.scheduleWithFixedDelay(this::pollInspections, OPERATION_POLL_MILLIS, OPERATION_POLL_MILLIS, TimeUnit.MILLISECONDS); + if (deployments != null) { + deploymentPolling = executor.scheduleWithFixedDelay(this::pollDeployments, + OPERATION_POLL_MILLIS, OPERATION_POLL_MILLIS, TimeUnit.MILLISECONDS); + } + } + + /** Deployment is a separate leased lane; its large I/O never runs on Bukkit's primary thread. */ + private void pollDeployments() { + if (closed || deployments == null || !registered || failures != 0 || !deploymentsAccepted + || !running.compareAndSet(false, true)) return; + CompletableFuture operation = new CompletableFuture<>(); + synchronized (operationLifecycle) { + if (closed) { + running.set(false); + return; + } + activeOperation = operation; + } + try { + claimAndDeploy(); + } catch (Exception failure) { + registered = false; + failures = Math.min(30, failures + 1); + if (failures == 1 || failures % 10 == 0) { + plugin.getLogger().warning("[Control] Bukkit deployment polling unavailable; VotingPlugin remains active"); + } + ScheduledFuture heartbeat = scheduled; + if (heartbeat != null) heartbeat.cancel(false); + if (!closed) schedule(Math.min(TimeUnit.MINUTES.toMillis(5), + 1000L << Math.min(failures - 1, 8))); + } finally { + operation.complete(null); + synchronized (operationLifecycle) { + if (activeOperation == operation) activeOperation = null; + } + running.set(false); + } + } + + private void claimAndDeploy() throws Exception { + JsonObject body = new JsonObject(); + body.addProperty("sessionId", sessionId.toString()); + Response response = send("POST", "/api/v1/nodes/" + settings.nodeId() + "/deployments", body); + if (response.status() == 204 || closed) return; + JsonObject claimed = requireObject(response, 200); + PluginDeploymentService.Task task = deploymentTask(claimed); + PluginDeploymentService.Result result = deployments.deploy(task, settings.endpoint(), directLocalDeploymentEndpoint, + settings.nodeId(), sessionId, credential, http, Duration.ofMillis(settings.requestTimeoutMillis()), + () -> !closed); + if (closed) return; + JsonObject submitted = new JsonObject(); + submitted.addProperty("sessionId", sessionId.toString()); + submitted.addProperty("success", result.success()); + submitted.addProperty("code", result.code()); + submitted.addProperty("message", boundedResultMessage(result.message())); + submitted.addProperty("attemptId", task.attemptId().toString()); + requireObject(send("POST", "/api/v1/nodes/" + settings.nodeId() + "/deployments/" + task.deploymentId() + + "/result", submitted), 200); + } + + static PluginDeploymentService.Task deploymentTask(JsonObject task) { + try { + if (task == null || task.size() != DEPLOYMENT_TASK_FIELDS.size() + || !DEPLOYMENT_TASK_FIELDS.containsAll(task.keySet())) throw new IllegalArgumentException(); + return new PluginDeploymentService.Task(UUID.fromString(string(task, "deploymentId")), + string(task, "artifactId"), string(task, "sha256"), Long.parseLong(string(task, "size")), + UUID.fromString(string(task, "attemptId"))); + } catch (RuntimeException failure) { + throw new IllegalArgumentException("deployment task is invalid"); + } } /** Polls the separately negotiated read-only lane on the connector worker. */ @@ -393,6 +485,8 @@ private void cycle() { && negotiatedCapability(node, "config.reward-files.v1", rewardFilesAccepted); boolean inspectionsWereAccepted = inspectionsAccepted; inspectionsAccepted = negotiatedCapability(node, "data.inspect.v1", inspectionsAccepted); + deploymentsAccepted = deployments != null + && negotiatedCapability(node, PluginDeploymentService.CAPABILITY, deploymentsAccepted); if (inspectionsAccepted && !inspectionsWereAccepted) { inspectionFailures = 0; inspectionRetryAtNanos = 0; @@ -445,6 +539,7 @@ private JsonObject register() throws Exception { voteSitesSyncAccepted = false; rewardFilesAccepted = false; inspectionsAccepted = false; + deploymentsAccepted = false; JsonObject body = sessionBody(); body.addProperty("nodeId", settings.nodeId()); body.addProperty("displayName", settings.nodeId()); @@ -455,13 +550,13 @@ private JsonObject register() throws Exception { .map(installed -> installed.getDescription().getName()).filter(name -> name != null && !name.isBlank()) .distinct().sorted(String.CASE_INSENSITIVE_ORDER).limit(128).forEach(detectedPlugins::add); body.add("detectedPlugins", detectedPlugins); - addCapabilities(body, configurations.supportsNamedRewardFiles()); + addCapabilities(body, configurations.supportsNamedRewardFiles(), deployments != null); return requireObject(send("POST", "/api/v1/nodes/register", body), 200, 201); } private JsonObject heartbeat() throws Exception { JsonObject body = sessionBody(); - addCapabilities(body, configurations.supportsNamedRewardFiles()); + addCapabilities(body, configurations.supportsNamedRewardFiles(), deployments != null); Response response = send("PUT", "/api/v1/nodes/" + settings.nodeId() + "/heartbeat", body); if (response.status() == 404) { registered = false; @@ -880,10 +975,15 @@ private JsonObject sessionBody() { return body; } - static void addCapabilities(JsonObject body, boolean rewardFilesSupported) { + static void addCapabilities(JsonObject body) { + addCapabilities(body, false, false); + } + + static void addCapabilities(JsonObject body, boolean rewardFilesSupported, boolean deploymentReady) { JsonArray capabilities = new JsonArray(); CAPABILITIES.stream().sorted().filter(capability -> rewardFilesSupported || !"config.reward-files.v1".equals(capability)).forEach(capabilities::add); + if (deploymentReady) capabilities.add(PluginDeploymentService.CAPABILITY); body.add("capabilities", capabilities); JsonArray required = new JsonArray(); required.add("config.files.v1"); @@ -961,6 +1061,9 @@ public void close() { if (polling != null) polling.cancel(false); ScheduledFuture inspection = inspectionPolling; if (inspection != null) inspection.cancel(false); + ScheduledFuture deployment = deploymentPolling; + if (deployment != null) deployment.cancel(false); + if (deployments != null) deployments.cancel(); inspectionExecutor.shutdownNow(); if (reload != null && Bukkit.isPrimaryThread()) reload.cancel(false); awaitShutdown(executor, operation); diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/control/PluginDeploymentService.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/control/PluginDeploymentService.java new file mode 100644 index 000000000..2b5256738 --- /dev/null +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/control/PluginDeploymentService.java @@ -0,0 +1,494 @@ +package com.bencodez.votingplugin.control; + +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.charset.CodingErrorAction; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; +import java.security.MessageDigest; +import java.time.Duration; +import java.util.HexFormat; +import java.util.Locale; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; +import java.util.jar.JarEntry; +import java.util.jar.JarFile; + +import com.bencodez.votingplugin.util.DurableFiles; + +/** + * Bounded, pull-only staging for the optional {@code plugin.deploy.v1} capability. + * It deliberately never reloads a plugin or starts a server. The small on-disk + * marker makes a leased task idempotent after a lost result acknowledgement. + */ +public final class PluginDeploymentService { + public static final String CAPABILITY = "plugin.deploy.v1"; + public static final long MAX_ARTIFACT_BYTES = 64L * 1024L * 1024L; + private static final int BUFFER_BYTES = 32 * 1024; + private static final int MAX_JAR_ENTRIES = 10_000; + private static final long MAX_INSPECTED_UNCOMPRESSED_BYTES = 256L * 1024L * 1024L; + private static final int MAX_PLUGIN_YML_BYTES = 64 * 1024; + private static final String MARKER = ".control-deployment"; + private static final ScheduledExecutorService BODY_DEADLINE_EXECUTOR = Executors.newSingleThreadScheduledExecutor(r -> { + Thread thread = new Thread(r, "VotingPlugin-control-artifact-deadline"); + thread.setDaemon(true); + return thread; + }); + + private final Path target; + private final Path root; + private final Path marker; + private final Path installedBackendJar; + private final boolean replaceExisting; + private final AtomicBoolean staging = new AtomicBoolean(); + private final AtomicReference activeResponse = new AtomicReference<>(); + + private PluginDeploymentService(Path target, Path installedBackendJar, boolean replaceExisting) throws IOException { + this.target = target.toAbsolutePath().normalize(); + this.installedBackendJar = installedBackendJar == null ? null : installedBackendJar.toAbsolutePath().normalize(); + this.replaceExisting = replaceExisting; + Path parent = this.target.getParent(); + if (parent == null) throw new IOException("deployment target has no parent"); + Files.createDirectories(parent); + this.root = parent.toRealPath(LinkOption.NOFOLLOW_LINKS); + if (!Files.isDirectory(root, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(root)) { + throw new IOException("deployment staging directory is unsafe"); + } + if (!this.target.startsWith(root) || Files.isSymbolicLink(this.target)) { + throw new IOException("deployment target escapes staging directory"); + } + if (Files.exists(this.target, LinkOption.NOFOLLOW_LINKS) + && !Files.isRegularFile(this.target, LinkOption.NOFOLLOW_LINKS)) { + throw new IOException("deployment target is not a regular file"); + } + this.marker = root.resolve(this.target.getFileName() + MARKER); + } + + /** Bukkit consumes an update only when its file name matches the currently loaded plugin JAR. */ + public static PluginDeploymentService backend(Path updateDirectory, Path currentPluginJar) throws IOException { + if (updateDirectory == null) throw new IOException("Bukkit update folder is unavailable"); + if (currentPluginJar == null || currentPluginJar.getFileName() == null + || !currentPluginJar.getFileName().toString().toLowerCase(Locale.ROOT).endsWith(".jar")) { + throw new IOException("Bukkit plugin JAR is unavailable"); + } + Path target = updateDirectory.resolve(currentPluginJar.getFileName().toString()).toAbsolutePath().normalize(); + Path installed = currentPluginJar.toAbsolutePath().normalize(); + // An empty/disabled Bukkit update folder can resolve to the loaded JAR's + // directory. Staging there would overwrite a file that Bukkit currently has open. + if (target.equals(installed) || Files.exists(target, LinkOption.NOFOLLOW_LINKS) + && Files.exists(installed, LinkOption.NOFOLLOW_LINKS) && Files.isSameFile(target, installed)) { + throw new IOException("Bukkit update folder resolves to the loaded plugin JAR"); + } + return new PluginDeploymentService(target, currentPluginJar, false); + } + + /** Proxies atomically replace their discovered plugin JAR only after a durable backup. */ + public static PluginDeploymentService proxy(Path currentPluginJar) throws IOException { + if (currentPluginJar == null || !currentPluginJar.getFileName().toString().endsWith(".jar")) { + throw new IOException("proxy plugin JAR is unavailable"); + } + if (!Files.isRegularFile(currentPluginJar, LinkOption.NOFOLLOW_LINKS) + || Files.isSymbolicLink(currentPluginJar)) throw new IOException("proxy plugin JAR is unsafe"); + return new PluginDeploymentService(currentPluginJar, null, true); + } + + public boolean isStaging() { return staging.get(); } + + /** Unblocks an in-progress response read during connector shutdown. */ + public void cancel() { + InputStream response = activeResponse.getAndSet(null); + if (response != null) { + try { response.close(); } catch (IOException ignored) { /* Shutdown is already in progress. */ } + } + } + + public Result deploy(Task task, URI endpoint, boolean directLocalHosted, String nodeId, UUID sessionId, + String credential, HttpClient http, Duration timeout, BooleanSupplier active) { + if (!staging.compareAndSet(false, true)) return Result.failure("DEPLOYMENT_FAILED", "Another deployment is still staging"); + try { + validate(task); + if (!credentialEndpointAllowed(endpoint, directLocalHosted)) { + return Result.failure("INSECURE_ENDPOINT", + "Verified update staging requires HTTPS unless Control is hosted directly on this node"); + } + if (!active.getAsBoolean()) return Result.failure("CANCELLED", "Deployment was cancelled before download"); + if (alreadyStaged(task)) return Result.restartRequired(); + URI artifact = endpoint.resolve("/api/v1/nodes/" + nodeId + "/deployments/" + task.deploymentId() + + "/artifact"); + HttpRequest request = HttpRequest.newBuilder(artifact).timeout(timeout) + .header("Authorization", "Bearer " + credential) + .header("X-Node-Session", sessionId.toString()) + .header("X-Deployment-Attempt", task.attemptId().toString()).GET().build(); + HttpResponse response = http.send(request, HttpResponse.BodyHandlers.ofInputStream()); + InputStream body = response.body(); + activeResponse.set(body); + ScheduledFuture bodyDeadline = BODY_DEADLINE_EXECUTOR.schedule(() -> { + try { body.close(); } catch (IOException ignored) { /* The staging operation reports the timeout. */ } + }, timeout.toMillis(), TimeUnit.MILLISECONDS); + try { + if (response.statusCode() != 200) { + body.close(); + return Result.failure("DOWNLOAD_FAILED", "Artifact download was rejected"); + } + String contentLength = response.headers().firstValue("Content-Length").orElse(null); + if (contentLength != null && (!contentLength.matches("[0-9]{1,9}") + || Long.parseLong(contentLength) != task.size())) { + body.close(); + return Result.failure("SIZE_MISMATCH", "Artifact size did not match the deployment task"); + } + if (!active.getAsBoolean()) { + body.close(); + return Result.failure("CANCELLED", "Deployment was cancelled before download"); + } + try (body) { + return stage(task, body, active); + } catch (IOException failure) { + if (!active.getAsBoolean()) return Result.failure("CANCELLED", "Deployment was cancelled before staging"); + return Result.failure("STAGING_FAILED", "Artifact could not be verified or staged on this node"); + } + } finally { + bodyDeadline.cancel(false); + activeResponse.compareAndSet(body, null); + } + } catch (InterruptedException failure) { + Thread.currentThread().interrupt(); + return Result.failure("CANCELLED", "Deployment download was interrupted"); + } catch (IOException | RuntimeException failure) { + return Result.failure("DOWNLOAD_FAILED", "Artifact download failed"); + } finally { + staging.set(false); + } + } + + /** Package-visible for deterministic artifact-validation tests. */ + Result stage(Task task, InputStream body, BooleanSupplier active) throws IOException { + validate(task); + if (!active.getAsBoolean()) return Result.failure("CANCELLED", "Deployment was cancelled before staging"); + if (alreadyStaged(task)) return Result.restartRequired(); + activeResponse.compareAndSet(null, body); + Path temporary = Files.createTempFile(root, target.getFileName().toString() + ".", ".download"); + Activation activation = null; + try { + MessageDigest digest = sha256(); + long written = copyExact(body, temporary, task.size(), digest, active); + force(temporary); + if (written != task.size()) return Result.failure("SIZE_MISMATCH", "Artifact size did not match the deployment task"); + if (!HexFormat.of().formatHex(digest.digest()).equals(task.sha256())) { + return Result.failure("HASH_MISMATCH", "Artifact digest did not match the deployment task"); + } + inspectJar(temporary); + if (!active.getAsBoolean()) return Result.failure("CANCELLED", "Deployment was cancelled before staging"); + activation = new Activation(); + prepareProxyBackup(); + writeMarker(task, activation); + activate(temporary, activation); + activation.discard(); + return Result.restartRequired(); + } catch (CancelledDeploymentException failure) { + return Result.failure("CANCELLED", "Deployment was cancelled before staging"); + } catch (InvalidArtifactException failure) { + return Result.failure("INVALID_ARTIFACT", "Artifact is not a bounded VotingPlugin JAR"); + } catch (IOException failure) { + if (activation != null) { + try { activation.rollback(); } + catch (IOException rollbackFailure) { failure.addSuppressed(rollbackFailure); } + } + if (!active.getAsBoolean()) return Result.failure("CANCELLED", "Deployment was cancelled before staging"); + throw failure; + } finally { + activeResponse.compareAndSet(body, null); + Files.deleteIfExists(temporary); + } + } + + private long copyExact(InputStream input, Path temporary, long expected, MessageDigest digest, + BooleanSupplier active) throws IOException { + long total = 0; + byte[] buffer = new byte[BUFFER_BYTES]; + try (var output = Files.newOutputStream(temporary, StandardOpenOption.TRUNCATE_EXISTING)) { + for (;;) { + if (!active.getAsBoolean()) throw new CancelledDeploymentException(); + int read = input.read(buffer); + if (read == -1) break; + total += read; + if (total > expected || total > MAX_ARTIFACT_BYTES) throw new InvalidArtifactException(); + digest.update(buffer, 0, read); + output.write(buffer, 0, read); + } + } + return total; + } + + private void inspectJar(Path artifact) throws IOException { + try (JarFile jar = new JarFile(artifact.toFile(), false)) { + int entries = 0; + long uncompressed = 0; + JarEntry pluginYml = null; + var iterator = jar.entries(); + while (iterator.hasMoreElements()) { + JarEntry entry = iterator.nextElement(); + if (++entries > MAX_JAR_ENTRIES || unsafeEntry(entry.getName())) throw new InvalidArtifactException(); + long size = entry.getSize(); + long compressed = entry.getCompressedSize(); + if (size < 0 || compressed < 0 || compressed > 0 && size > compressed * 200L + || (uncompressed += size) > MAX_INSPECTED_UNCOMPRESSED_BYTES) { + throw new InvalidArtifactException(); + } + if ("plugin.yml".equals(entry.getName())) { + if (pluginYml != null || size > MAX_PLUGIN_YML_BYTES) throw new InvalidArtifactException(); + pluginYml = entry; + } + } + if (pluginYml == null || !isVotingPluginYml(jar, pluginYml)) throw new InvalidArtifactException(); + } catch (java.util.zip.ZipException failure) { + throw new InvalidArtifactException(); + } + } + + private static boolean unsafeEntry(String name) { + if (name == null || name.isEmpty() || name.length() > 512 || name.startsWith("/") + || name.startsWith("\\") || name.contains("\\") || name.indexOf('\0') >= 0) return true; + for (String component : name.split("/", -1)) { + if (component.equals(".") || component.equals("..")) return true; + } + return false; + } + + private static boolean isVotingPluginYml(JarFile jar, JarEntry entry) throws IOException { + byte[] bytes; + try (InputStream input = jar.getInputStream(entry)) { bytes = input.readNBytes(MAX_PLUGIN_YML_BYTES + 1); } + if (bytes.length > MAX_PLUGIN_YML_BYTES) return false; + String yml; + try { + yml = StandardCharsets.UTF_8.newDecoder().onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT).decode(java.nio.ByteBuffer.wrap(bytes)).toString(); + } catch (java.nio.charset.CharacterCodingException failure) { + return false; + } + boolean found = false; + for (String line : yml.split("\\R")) { + if (!line.startsWith("name:")) continue; + if (found) return false; + found = true; + String value = line.substring(5).trim(); + int comment = value.indexOf('#'); + if (comment >= 0) value = value.substring(0, comment).trim(); + if (!"VotingPlugin".equals(unquote(value))) return false; + } + return found; + } + + private static String unquote(String value) { + return value.length() >= 2 && ((value.startsWith("\"") && value.endsWith("\"")) + || (value.startsWith("'") && value.endsWith("'"))) ? value.substring(1, value.length() - 1) : value; + } + + private void prepareProxyBackup() throws IOException { + if (replaceExisting) { + if (!Files.isRegularFile(target, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(target)) { + throw new IOException("current proxy plugin JAR is unsafe"); + } + Path backup = root.resolve(target.getFileName() + ".control-backup"); + if (Files.exists(backup, LinkOption.NOFOLLOW_LINKS) + && (!Files.isRegularFile(backup, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(backup))) { + throw new IOException("proxy plugin backup is unsafe"); + } + Path backupTemp = Files.createTempFile(root, target.getFileName().toString() + ".", ".backup"); + try { + Files.copy(target, backupTemp, StandardCopyOption.REPLACE_EXISTING); + force(backupTemp); + move(backupTemp, backup); + forceDirectory(root); + } finally { Files.deleteIfExists(backupTemp); } + } + } + + private void activate(Path temporary, Activation activation) throws IOException { + move(temporary, target); + activation.published = true; + force(target); + forceDirectory(root); + } + + /** Holds a private copy until the durable idempotency marker has been published. */ + private final class Activation { + private final Path previous; + private boolean published; + private boolean markerPublished; + + private Activation() throws IOException { + if (!Files.exists(target, LinkOption.NOFOLLOW_LINKS)) { + previous = null; + return; + } + if (!Files.isRegularFile(target, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(target)) { + throw new IOException("deployment target is unsafe"); + } + previous = Files.createTempFile(root, target.getFileName().toString() + ".", ".rollback"); + Files.copy(target, previous, StandardCopyOption.REPLACE_EXISTING); + force(previous); + } + + private void rollback() throws IOException { + if (markerPublished) DurableFiles.deleteIfExists(marker); + if (!published) { + discard(); + return; + } + if (previous == null) { + if (!Files.isRegularFile(target, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(target)) { + throw new IOException("deployed target cannot be safely removed"); + } + Files.delete(target); + } else { + move(previous, target); + force(target); + } + forceDirectory(root); + discard(); + } + + private void discard() { + if (previous == null) return; + try { Files.deleteIfExists(previous); } + catch (IOException ignored) { /* A private stale rollback copy is safer than a false deployment result. */ } + } + } + + private void writeMarker(Task task) throws IOException { + writeMarker(task, null); + } + + private void writeMarker(Task task, Activation activation) throws IOException { + Path temporary = Files.createTempFile(root, target.getFileName().toString() + ".", ".marker"); + try { + Files.writeString(temporary, task.deploymentId() + "\n" + task.sha256() + "\n" + task.size() + "\n", + StandardCharsets.US_ASCII, StandardOpenOption.TRUNCATE_EXISTING); + force(temporary); + move(temporary, marker); + if (activation != null) activation.markerPublished = true; + forceDirectory(root); + } finally { Files.deleteIfExists(temporary); } + } + + private boolean alreadyStaged(Task task) throws IOException { + if (!Files.isRegularFile(marker, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(marker) + || Files.size(marker) > 256) return false; + if (replaceExisting) { + Path backup = root.resolve(target.getFileName() + ".control-backup"); + if (!Files.isRegularFile(backup, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(backup)) return false; + } + String[] fields = Files.readString(marker, StandardCharsets.US_ASCII).split("\\R", -1); + // A lost acknowledgement can cause Control to issue a new deployment ID + // for the same verified artifact after reconnect or restart. The marker + // only proves a completed prior stage; the target is checked below by + // exact size, digest and plugin identity before acknowledging the retry. + if (fields.length < 3 || !task.sha256().equals(fields[1]) + || !Long.toString(task.size()).equals(fields[2])) return false; + if (!replaceExisting && !Files.exists(target, LinkOption.NOFOLLOW_LINKS)) { + // Bukkit removes the staged update JAR after consuming it on restart. A + // missing staging file alone is not proof: it may have been deleted or + // quarantined. Confirm the installed backend JAR is the exact artifact. + return installedBackendJar != null && fileMatches(installedBackendJar, task); + } + return targetMatches(task); + } + + private boolean targetMatches(Task task) throws IOException { + return fileMatches(target, task); + } + + private boolean fileMatches(Path candidate, Task task) throws IOException { + if (!Files.isRegularFile(candidate, LinkOption.NOFOLLOW_LINKS) || Files.isSymbolicLink(candidate) + || Files.size(candidate) != task.size()) return false; + MessageDigest digest = sha256(); + try (InputStream input = Files.newInputStream(candidate, StandardOpenOption.READ, LinkOption.NOFOLLOW_LINKS)) { + byte[] bytes = new byte[BUFFER_BYTES]; + for (int read; (read = input.read(bytes)) != -1;) digest.update(bytes, 0, read); + } + if (!task.sha256().equals(HexFormat.of().formatHex(digest.digest()))) return false; + try { + inspectJar(candidate); + return true; + } catch (InvalidArtifactException invalid) { + return false; + } + } + + static void move(Path source, Path destination) throws IOException { + // The proxy target is executable on its next startup. A copy/delete fallback + // can leave it missing or partial after a crash, even with a durable backup. + Files.move(source, destination, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } + + private static void force(Path file) throws IOException { + try (FileChannel channel = FileChannel.open(file, StandardOpenOption.WRITE)) { channel.force(true); } + } + + private static void forceDirectory(Path directory) throws IOException { + DurableFiles.forceDirectory(directory); + } + + /** True when a deployment bearer credential may be sent to this Control endpoint. */ + public static boolean credentialEndpointAllowed(URI endpoint, boolean directLocalHosted) { + if (endpoint == null) return false; + if ("https".equalsIgnoreCase(endpoint.getScheme())) return true; + return directLocalHosted && "http".equalsIgnoreCase(endpoint.getScheme()) + && isLoopbackHost(endpoint.getHost()); + } + + private static boolean isLoopbackHost(String host) { + if (host == null) return false; + String normalized = host; + if (normalized.length() >= 2 && normalized.charAt(0) == '[' + && normalized.charAt(normalized.length() - 1) == ']') { + normalized = normalized.substring(1, normalized.length() - 1); + } + if ("localhost".equalsIgnoreCase(normalized) || "::1".equalsIgnoreCase(normalized) + || "0:0:0:0:0:0:0:1".equalsIgnoreCase(normalized)) return true; + String[] octets = normalized.split("\\.", -1); + if (octets.length != 4 || !"127".equals(octets[0])) return false; + for (int index = 1; index < octets.length; index++) { + if (!octets[index].matches("[0-9]{1,3}") || Integer.parseInt(octets[index]) > 255) return false; + } + return true; + } + + private static MessageDigest sha256() { + try { return MessageDigest.getInstance("SHA-256"); } + catch (java.security.NoSuchAlgorithmException failure) { throw new IllegalStateException("SHA-256 is unavailable", failure); } + } + + private static void validate(Task task) { + if (task == null || task.deploymentId() == null || task.attemptId() == null || task.artifactId() == null + || !task.artifactId().matches("[A-Za-z0-9][A-Za-z0-9._-]{0,127}") || task.sha256() == null + || !task.sha256().matches("[0-9a-fA-F]{64}") || task.size() < 1 || task.size() > MAX_ARTIFACT_BYTES) { + throw new IllegalArgumentException("deployment task is invalid"); + } + } + + public record Task(UUID deploymentId, String artifactId, String sha256, long size, UUID attemptId) { + public Task { if (sha256 != null) sha256 = sha256.toLowerCase(Locale.ROOT); } + } + public record Result(boolean success, String code, String message) { + static Result restartRequired() { return new Result(true, "RESTART_REQUIRED", "Plugin update staged; restart is required"); } + static Result failure(String code, String message) { return new Result(false, code, message); } + } + private static final class InvalidArtifactException extends IOException { private static final long serialVersionUID = 1L; } + private static final class CancelledDeploymentException extends IOException { private static final long serialVersionUID = 1L; } +} diff --git a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/control/ControlConnector.java b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/control/ControlConnector.java index aa1f4fa40..a0576e7be 100644 --- a/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/control/ControlConnector.java +++ b/VotingPlugin/src/main/java/com/bencodez/votingplugin/proxy/control/ControlConnector.java @@ -2,6 +2,7 @@ import java.io.IOException; import java.net.URI; +import java.net.URISyntaxException; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; @@ -22,6 +23,8 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadLocalRandom; @@ -33,13 +36,16 @@ import java.util.function.Supplier; import java.util.regex.Pattern; +import com.bencodez.votingplugin.control.PluginDeploymentService; import com.bencodez.votingplugin.proxy.VotingPluginProxy; +import com.bencodez.votingplugin.proxy.control.HostedControlManager.HostConfiguration; import com.bencodez.votingplugin.proxy.VotingPluginProxyConfig; import com.bencodez.votingplugin.proxy.presence.BackendPresenceStatus; import com.bencodez.votingplugin.proxy.control.ProxyControlResultStore.Route; import com.bencodez.votingplugin.proxy.control.ProxyControlResultStore.StoredResult; import com.bencodez.votingplugin.util.BoundedHttpBodyHandler; import com.bencodez.votingplugin.util.ControlCredentialFile; +import com.bencodez.votingplugin.util.DurableFiles; import com.google.gson.JsonArray; import com.google.gson.JsonElement; import com.google.gson.JsonObject; @@ -63,6 +69,7 @@ public final class ControlConnector implements AutoCloseable { private static final String PROXY_METHOD_PRESET = "proxy-method"; private static final String INTERNAL_OPERATION_TYPE = "_controlOperationType"; private static final String INTERNAL_REQUIRED_CAPABILITY = "_controlRequiredCapability"; + private static final Set DEPLOYMENT_TASK_FIELDS = Set.of("deploymentId", "artifactId", "sha256", "size", "attemptId"); private static final long OPERATION_POLL_MILLIS = 1000; private static final long MAX_BACKOFF_MILLIS = TimeUnit.MINUTES.toMillis(5); private static final long OPERATION_SHUTDOWN_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(65); @@ -77,6 +84,11 @@ public final class ControlConnector implements AutoCloseable { private final ProxyRoutingConfigurationService configurationService; private final ProxyMethodConfigurationService methodConfigurationService; private final ProxyConfigurationFileService fileConfigurationService; + private final PluginDeploymentService deployments; + private final HttpClient deploymentHttp; + private final String deploymentCredential; + private final boolean directLocalDeploymentEndpoint; + private final ExecutorService deploymentExecutor; private final Function> communicationTest; private final Runnable runtimeReplacement; private final Path dataDirectory; @@ -91,12 +103,17 @@ public final class ControlConnector implements AutoCloseable { private volatile boolean closed; private volatile boolean registered; private volatile boolean configurationAccepted; + private volatile boolean deploymentsAccepted; private volatile Set acceptedCapabilities = Set.of(); private volatile int failures; private volatile long snapshotSequence; private volatile ScheduledFuture scheduled; private volatile ScheduledFuture operationPolling; + private volatile ScheduledFuture deploymentPolling; private volatile CompletableFuture activeRequest; + /* The deployment executor may discard queued AsyncSupply work during shutdown. + * Retain its future so close() can complete the dependent operation chain. */ + private volatile CompletableFuture activeDeploymentWork; private volatile CompletableFuture activeOperation; private volatile Status status = Status.STARTING; @@ -130,6 +147,19 @@ private ControlConnector(Settings settings, ScheduledExecutorService scheduler, Function> communicationTest, ProxyMethodConfigurationService methodConfigurationService, Runnable runtimeReplacement, ProxyConfigurationFileService fileConfigurationService) { + this(settings, scheduler, transport, snapshotSource, logger, sessionId, jitterSource, configurationService, + dataDirectory, route, recovering, recoveryComplete, communicationTest, methodConfigurationService, + runtimeReplacement, fileConfigurationService, null, null, null, false); + } + + private ControlConnector(Settings settings, ScheduledExecutorService scheduler, Transport transport, + Supplier> snapshotSource, Consumer logger, UUID sessionId, + LongSupplier jitterSource, ProxyRoutingConfigurationService configurationService, Path dataDirectory, + Route route, boolean recovering, Runnable recoveryComplete, + Function> communicationTest, + ProxyMethodConfigurationService methodConfigurationService, Runnable runtimeReplacement, + ProxyConfigurationFileService fileConfigurationService, PluginDeploymentService deployments, + HttpClient deploymentHttp, String deploymentCredential, boolean directLocalDeploymentEndpoint) { this.settings = Objects.requireNonNull(settings, "settings"); this.scheduler = Objects.requireNonNull(scheduler, "scheduler"); this.transport = Objects.requireNonNull(transport, "transport"); @@ -142,6 +172,15 @@ private ControlConnector(Settings settings, ScheduledExecutorService scheduler, this.communicationTest = communicationTest; this.runtimeReplacement = runtimeReplacement; this.fileConfigurationService = fileConfigurationService; + this.deployments = deployments; + this.deploymentHttp = deploymentHttp; + this.deploymentCredential = deploymentCredential; + this.directLocalDeploymentEndpoint = directLocalDeploymentEndpoint; + this.deploymentExecutor = deployments == null ? null : Executors.newSingleThreadExecutor(runnable -> { + Thread thread = new Thread(runnable, "votingplugin-control-proxy-deployment"); + thread.setDaemon(true); + return thread; + }); this.dataDirectory = dataDirectory; this.route = route; this.recovering = recovering; @@ -192,6 +231,7 @@ public static ControlConnector create(VotingPluginProxy proxy) throws IOExceptio } } boolean recovering = recovered != null && recovered.routeRequired(); + boolean deploymentRouteCurrent = config.getControlEnabled() && !recovering; String credential = ControlCredentialFile.read(dataDirectory, credentialName); HttpControlTransport transport = new HttpControlTransport(settings.endpoint(), credential, settings.connectTimeoutMillis(), settings.requestTimeoutMillis()); @@ -210,12 +250,31 @@ public static ControlConnector create(VotingPluginProxy proxy) throws IOExceptio backends.sort(Comparator.comparing(ObservedBackend::backendId)); return List.copyOf(backends); }; + HostConfiguration hosted = new HostConfiguration(config.getControlHostedEnabled(), + config.getControlHostedAutoDownload(), config.getControlHostedAutoUpdate(), + config.getControlHostedDownloadUrl(), config.getControlHostedSha256(), + config.getControlHostedJarFile(), config.getControlHostedDataDirectory(), + config.getControlHostedHost(), config.getControlHostedPort(), + config.getControlHostedStartupTimeoutSeconds(), config.getControlHostedDownloadTimeoutSeconds()); + boolean directLocalDeploymentEndpoint = HostedControlManager.isDirectLocalEndpoint( + settings.endpoint().toString(), hosted); + boolean deploymentEndpointAllowed = PluginDeploymentService.credentialEndpointAllowed( + settings.endpoint(), directLocalDeploymentEndpoint); + PluginDeploymentService deployments = deploymentRouteCurrent && deploymentEndpointAllowed + ? prepareDeployment(proxy) : null; + if (deploymentRouteCurrent && !deploymentEndpointAllowed) { + proxy.log("[Control] Plugin deployment staging requires HTTPS unless Control is hosted directly on this node"); + } + HttpClient deploymentHttp = deployments == null ? null : HttpClient.newBuilder() + .connectTimeout(Duration.ofMillis(settings.connectTimeoutMillis())) + .followRedirects(HttpClient.Redirect.NEVER).build(); ControlConnector connector = new ControlConnector(settings, proxy.getScheduler(), transport, snapshot, message -> proxy.log("[Control] " + message), UUID.randomUUID(), () -> ThreadLocalRandom.current().nextLong(), new ProxyRoutingConfigurationService(proxy), dataDirectory, route, recovering, proxy::restartControlServicesAfterRecovery, server -> proxy.testBackendCommunication(server, 5000L), new ProxyMethodConfigurationService(proxy), - () -> proxy.reloadCore(true), new ProxyConfigurationFileService(proxy)); + () -> proxy.reloadCore(true), new ProxyConfigurationFileService(proxy), + deployments, deploymentHttp, credential, directLocalDeploymentEndpoint); if (recovered != null) connector.completedTasks.putAll(recovered.results()); return connector; } @@ -228,6 +287,116 @@ public void start() { schedule(0); operationPolling = scheduler.scheduleWithFixedDelay(this::pollOperations, OPERATION_POLL_MILLIS, OPERATION_POLL_MILLIS, TimeUnit.MILLISECONDS); + if (deployments != null) { + deploymentPolling = scheduler.scheduleWithFixedDelay(this::pollDeployments, + OPERATION_POLL_MILLIS, OPERATION_POLL_MILLIS, TimeUnit.MILLISECONDS); + } + } + + /** Keeps large artifact transfer and disk staging off proxy event threads and serializes it with operations. */ + private void pollDeployments() { + CompletableFuture done = new CompletableFuture<>(); + synchronized (operationLifecycle) { + if (closed || !registered || status != Status.CONNECTED || !deploymentsAccepted || deployments == null + || !inFlight.compareAndSet(false, true)) return; + activeOperation = done; + } + CompletableFuture work; + try { + CompletableFuture claim = transport.send(deploymentClaimRequest()); + activeRequest = claim; + work = claim.thenCompose(this::handleDeploymentClaim); + } catch (RuntimeException failure) { + work = CompletableFuture.failedFuture(failure); + } + work.whenComplete((ignored, failure) -> { + Throwable cause = failure == null ? null : unwrap(failure); + activeRequest = null; + if (cause == null) { + done.complete(null); + } else { + registered = false; + done.completeExceptionally(cause); + } + if (activeOperation == done) activeOperation = null; + finishCycle(); + if (cause != null && !closed) { + ScheduledFuture heartbeat = scheduled; + if (heartbeat != null) heartbeat.cancel(false); + onFailure(cause); + } + }); + } + + private Request deploymentClaimRequest() { + JsonObject body = sessionBody(); + return new Request("POST", "/api/v1/nodes/" + settings.nodeId() + "/deployments", body.toString()); + } + + private CompletableFuture handleDeploymentClaim(Response response) { + if (response.statusCode == 204 || closed) return CompletableFuture.completedFuture(null); + if (response.statusCode == 404) { + registered = false; + throw new RegistryLostException(); + } + requireSuccess(response); + PluginDeploymentService.Task task = deploymentTask(parseObject(response.body)); + final CompletableFuture deploymentWork; + synchronized (operationLifecycle) { + if (closed) return CompletableFuture.completedFuture(null); + deploymentWork = CompletableFuture.supplyAsync(() -> deployments.deploy(task, settings.endpoint(), + directLocalDeploymentEndpoint, settings.nodeId(), sessionId, deploymentCredential, deploymentHttp, + Duration.ofMillis(settings.requestTimeoutMillis()), () -> !closed), deploymentExecutor); + activeDeploymentWork = deploymentWork; + } + return deploymentWork.thenCompose(result -> { + if (closed) return CompletableFuture.completedFuture(null); + JsonObject body = new JsonObject(); + body.addProperty("sessionId", sessionId.toString()); + body.addProperty("success", result.success()); + body.addProperty("code", result.code()); + body.addProperty("message", boundedResultMessage(result.message())); + body.addProperty("attemptId", task.attemptId().toString()); + CompletableFuture submitted = transport.send(new Request("POST", "/api/v1/nodes/" + + settings.nodeId() + "/deployments/" + task.deploymentId() + "/result", body.toString())); + activeRequest = submitted; + return submitted.thenAccept(ControlConnector::requireSuccess); + }).whenComplete((ignored, failure) -> { + synchronized (operationLifecycle) { + if (activeDeploymentWork == deploymentWork) activeDeploymentWork = null; + } + }); + } + + static PluginDeploymentService.Task deploymentTask(JsonObject task) { + try { + if (task == null || task.size() != DEPLOYMENT_TASK_FIELDS.size() + || !DEPLOYMENT_TASK_FIELDS.containsAll(task.keySet())) throw new IllegalArgumentException(); + return new PluginDeploymentService.Task(UUID.fromString(requireString(task, "deploymentId")), + requireString(task, "artifactId"), requireString(task, "sha256"), + Long.parseLong(requireString(task, "size")), UUID.fromString(requireString(task, "attemptId"))); + } catch (RuntimeException failure) { + throw new MalformedResponseException(); + } + } + + private static PluginDeploymentService prepareDeployment(VotingPluginProxy proxy) { + if (!proxyDeploymentSupported(System.getProperty("os.name", ""))) { + proxy.log("[Control] Plugin deployment staging is unavailable on Windows proxies; capability not advertised"); + return null; + } + try { + var source = proxy.getClass().getProtectionDomain().getCodeSource(); + if (source == null || !"file".equalsIgnoreCase(source.getLocation().getProtocol())) return null; + return PluginDeploymentService.proxy(Path.of(source.getLocation().toURI())); + } catch (IOException | URISyntaxException | RuntimeException failure) { + proxy.log("[Control] Plugin deployment staging is unavailable; capability not advertised"); + return null; + } + } + + static boolean proxyDeploymentSupported(String osName) { + return !DurableFiles.isWindowsName(osName); } /** Polls only the operation queue; heartbeat and presence retain their configured cadence. */ @@ -435,6 +604,8 @@ private void handlePrimaryResponse(Response response, boolean registration) { acceptedCapabilities = Set.copyOf(negotiated); configurationAccepted = acceptedCapabilities.stream().anyMatch(Set.of(CONFIGURATION_CAPABILITY, COMMUNICATION_TEST_CAPABILITY, PROXY_METHOD_CAPABILITY, PROXY_FILE_CAPABILITY)::contains); + deploymentsAccepted = deployments != null + && acceptedCapabilities.contains(PluginDeploymentService.CAPABILITY); } } @@ -1088,12 +1259,19 @@ static List boundedResultChanges(List changes) { } private void addCapabilities(JsonObject body) { + addCapabilities(body, configurationService != null, communicationTest != null, methodConfigurationService != null, + fileConfigurationService != null, deployments != null); + } + + static void addCapabilities(JsonObject body, boolean configurationReady, boolean communicationReady, + boolean methodReady, boolean fileReady, boolean deploymentReady) { JsonArray advertised = new JsonArray(); BASE_CAPABILITIES.stream().sorted().forEach(advertised::add); - if (configurationService != null) advertised.add(CONFIGURATION_CAPABILITY); - if (communicationTest != null) advertised.add(COMMUNICATION_TEST_CAPABILITY); - if (methodConfigurationService != null) advertised.add(PROXY_METHOD_CAPABILITY); - if (fileConfigurationService != null) advertised.add(PROXY_FILE_CAPABILITY); + if (configurationReady) advertised.add(CONFIGURATION_CAPABILITY); + if (communicationReady) advertised.add(COMMUNICATION_TEST_CAPABILITY); + if (methodReady) advertised.add(PROXY_METHOD_CAPABILITY); + if (fileReady) advertised.add(PROXY_FILE_CAPABILITY); + if (deploymentReady) advertised.add(PluginDeploymentService.CAPABILITY); body.add("capabilities", advertised); JsonArray required = new JsonArray(); required.add("presence.snapshot"); @@ -1133,6 +1311,12 @@ public void close() { } ScheduledFuture polling = operationPolling; if (polling != null) polling.cancel(false); + ScheduledFuture deployment = deploymentPolling; + if (deployment != null) deployment.cancel(false); + if (deployments != null) deployments.cancel(); + CompletableFuture deploymentWork = activeDeploymentWork; + if (deploymentWork != null) deploymentWork.cancel(true); + if (deploymentExecutor != null) deploymentExecutor.shutdownNow(); CompletableFuture request = activeRequest; if (request != null) { request.cancel(true); @@ -1150,6 +1334,16 @@ public void close() { throw new IllegalStateException("Interrupted while waiting for the Control operation", e); } } + if (deploymentExecutor != null) { + try { + if (!deploymentExecutor.awaitTermination(OPERATION_SHUTDOWN_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) { + throw new IllegalStateException("Control deployment worker did not stop before connector shutdown"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while waiting for the Control deployment worker", e); + } + } transport.close(); } diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/control/BackendControlConnectorProtocolTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/control/BackendControlConnectorProtocolTest.java index 11c607db9..15e655cec 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/control/BackendControlConnectorProtocolTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/control/BackendControlConnectorProtocolTest.java @@ -41,6 +41,22 @@ class BackendControlConnectorProtocolTest { "{\"error\":{\"code\":\"NODE_NOT_FOUND\"}}"))); } + @Test void deploymentTaskRejectsUnknownV1Fields() { + JsonObject task = deploymentTask(); + task.addProperty("unexpected", "value"); + assertThrows(IllegalArgumentException.class, () -> BackendControlConnector.deploymentTask(task)); + } + + private static JsonObject deploymentTask() { + JsonObject task = new JsonObject(); + task.addProperty("deploymentId", "00000000-0000-0000-0000-000000000001"); + task.addProperty("artifactId", "VotingPlugin.jar"); + task.addProperty("sha256", "a".repeat(64)); + task.addProperty("size", "1"); + task.addProperty("attemptId", "00000000-0000-0000-0000-000000000002"); + return task; + } + @Test void abandonedBackendIntentBecomesATerminalRecoveryResult() { JsonObject anticipated = new JsonObject(); anticipated.addProperty("attemptId", "00000000-0000-0000-0000-000000000199"); @@ -207,7 +223,7 @@ class BackendControlConnectorProtocolTest { @Test void registrationAdvertisesCommentPreservingFilesAsAnOptionalCapability() { JsonObject registration = new JsonObject(); - BackendControlConnector.addCapabilities(registration, true); + BackendControlConnector.addCapabilities(registration, true, false); JsonArray advertised = registration.getAsJsonArray("capabilities"); assertTrue(advertised.asList().stream() @@ -229,13 +245,29 @@ class BackendControlConnectorProtocolTest { .anyMatch(value -> "data.inspect.v1".equals(value.getAsString()))); } + @Test void deploymentCapabilityIsOnlyAddedWhenStagingWasPrepared() { + JsonObject unavailable = new JsonObject(); + BackendControlConnector.addCapabilities(unavailable, true, false); + assertFalse(unavailable.getAsJsonArray("capabilities").asList().stream() + .anyMatch(value -> PluginDeploymentService.CAPABILITY.equals(value.getAsString()))); + + JsonObject ready = new JsonObject(); + BackendControlConnector.addCapabilities(ready, true, true); + assertTrue(ready.getAsJsonArray("capabilities").asList().stream() + .anyMatch(value -> PluginDeploymentService.CAPABILITY.equals(value.getAsString()))); + assertTrue(ready.getAsJsonArray("capabilities").asList().stream() + .anyMatch(value -> "config.reward-files.v1".equals(value.getAsString()))); + } + @Test void unsupportedFilesystemDoesNotAdvertiseNamedRewards() { JsonObject registration = new JsonObject(); - BackendControlConnector.addCapabilities(registration, false); + BackendControlConnector.addCapabilities(registration, false, true); assertFalse(registration.getAsJsonArray("capabilities").asList().stream() .anyMatch(value -> "config.reward-files.v1".equals(value.getAsString()))); assertTrue(registration.getAsJsonArray("capabilities").asList().stream() .anyMatch(value -> "config.files.v1".equals(value.getAsString()))); + assertTrue(registration.getAsJsonArray("capabilities").asList().stream() + .anyMatch(value -> PluginDeploymentService.CAPABILITY.equals(value.getAsString()))); } @Test void heartbeatRetainsOmittedCapabilitiesAndHonorsExplicitReplacement() { diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/control/PluginDeploymentServiceTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/control/PluginDeploymentServiceTest.java new file mode 100644 index 000000000..423e46db8 --- /dev/null +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/control/PluginDeploymentServiceTest.java @@ -0,0 +1,379 @@ +package com.bencodez.votingplugin.control; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.Arrays; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.FileSystems; +import java.nio.file.AtomicMoveNotSupportedException; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.util.HexFormat; +import java.util.UUID; +import java.util.jar.JarEntry; +import java.util.jar.JarOutputStream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class PluginDeploymentServiceTest { + @TempDir Path directory; + + @Test void publicationRefusesAProviderWithoutAtomicMoves() throws Exception { + Path source = directory.resolve("VotingPlugin.jar"); + byte[] original = jar("name: VotingPlugin\nversion: old\n"); + Files.write(source, original); + Path archive = directory.resolve("non-atomic.zip"); + try (var zip = FileSystems.newFileSystem(java.net.URI.create("jar:" + archive.toUri()), + java.util.Map.of("create", "true"))) { + Path destination = zip.getPath("/VotingPlugin.jar"); + assertThrows(AtomicMoveNotSupportedException.class, + () -> PluginDeploymentService.move(source, destination)); + assertArrayEquals(original, Files.readAllBytes(source)); + assertFalse(Files.exists(destination)); + } + } + + @Test void backendStagesOnlyAnExactVerifiedVotingPluginJarAndIsIdempotent() throws Exception { + byte[] artifact = jar("name: VotingPlugin\nmain: example.Main\n"); + PluginDeploymentService service = PluginDeploymentService.backend(directory.resolve("update"), Path.of("VotingPlugin.jar")); + PluginDeploymentService.Task task = task(artifact); + + PluginDeploymentService.Result staged = service.stage(task, new ByteArrayInputStream(artifact), () -> true); + + assertTrue(staged.success()); + assertEquals("RESTART_REQUIRED", staged.code()); + assertEquals(artifact.length, Files.size(directory.resolve("update/VotingPlugin.jar"))); + assertTrue(service.stage(task, new ByteArrayInputStream(new byte[0]), () -> true).success()); + } + + @Test void backendStagesUsingTheInstalledPluginJarFileName() throws Exception { + byte[] artifact = jar("name: VotingPlugin\n"); + Path update = directory.resolve("update"); + PluginDeploymentService service = PluginDeploymentService.backend(update, + directory.resolve("plugins/VotingPlugin-7.1.2.jar")); + + assertEquals("RESTART_REQUIRED", service.stage(task(artifact), new ByteArrayInputStream(artifact), () -> true).code()); + assertArrayEquals(artifact, Files.readAllBytes(update.resolve("VotingPlugin-7.1.2.jar"))); + assertFalse(Files.exists(update.resolve("VotingPlugin.jar"))); + } + + @Test void backendRejectsAnEmptyOrDisabledBukkitUpdateFolderThatResolvesToTheLoadedJar() { + assertThrows(IOException.class, + () -> PluginDeploymentService.backend(Path.of(""), Path.of("VotingPlugin.jar"))); + } + + @Test void backendMatchingMarkerSurvivesBukkitConsumingTheStagedJar() throws Exception { + byte[] artifact = jar("name: VotingPlugin\nversion: candidate\n"); + Path update = directory.resolve("update"); + Path installed = directory.resolve("VotingPlugin.jar"); + Files.write(installed, jar("name: VotingPlugin\nversion: old\n")); + PluginDeploymentService service = PluginDeploymentService.backend(update, installed); + PluginDeploymentService.Task task = task(artifact); + + assertTrue(service.stage(task, new ByteArrayInputStream(artifact), () -> true).success()); + Files.write(installed, artifact); + Files.delete(update.resolve("VotingPlugin.jar")); + + assertTrue(service.stage(task, new ByteArrayInputStream(new byte[0]), () -> true).success()); + assertFalse(Files.exists(update.resolve("VotingPlugin.jar")), + "a lost acknowledgement after restart must recognize the artifact Bukkit already consumed"); + } + + @Test void retryWithNewDeploymentIdAcknowledgesOnlyTheSameVerifiedArtifact() throws Exception { + byte[] artifact = jar("name: VotingPlugin\nversion: candidate\n"); + Path update = directory.resolve("update"); + Path installed = directory.resolve("VotingPlugin.jar"); + Files.write(installed, jar("name: VotingPlugin\nversion: old\n")); + PluginDeploymentService service = PluginDeploymentService.backend(update, installed); + PluginDeploymentService.Task original = task(artifact); + PluginDeploymentService.Task retry = task(artifact); + assertTrue(service.stage(original, new ByteArrayInputStream(artifact), () -> true).success()); + assertFalse(original.deploymentId().equals(retry.deploymentId())); + + assertEquals("RESTART_REQUIRED", + service.stage(retry, new ByteArrayInputStream(new byte[0]), () -> true).code()); + assertArrayEquals(artifact, Files.readAllBytes(update.resolve("VotingPlugin.jar"))); + + Files.write(installed, artifact); + Files.delete(update.resolve("VotingPlugin.jar")); + assertEquals("RESTART_REQUIRED", + service.stage(retry, new ByteArrayInputStream(new byte[0]), () -> true).code()); + assertFalse(Files.exists(update.resolve("VotingPlugin.jar"))); + } + + @Test void backendRestagesWhenTheUpdateJarDisappearsBeforeBukkitConsumesIt() throws Exception { + byte[] artifact = jar("name: VotingPlugin\nversion: candidate\n"); + Path update = directory.resolve("update"); + Path installed = directory.resolve("VotingPlugin.jar"); + Files.write(installed, jar("name: VotingPlugin\nversion: old\n")); + PluginDeploymentService service = PluginDeploymentService.backend(update, installed); + PluginDeploymentService.Task task = task(artifact); + + assertTrue(service.stage(task, new ByteArrayInputStream(artifact), () -> true).success()); + Files.delete(update.resolve("VotingPlugin.jar")); + + assertTrue(service.stage(task, new ByteArrayInputStream(artifact), () -> true).success()); + assertArrayEquals(artifact, Files.readAllBytes(update.resolve("VotingPlugin.jar")), + "a deleted or quarantined update must be staged again before restart"); + } + + @Test void credentialedDeploymentRequiresHttpsUnlessSameNodeHostedHttpWasProven() { + assertTrue(PluginDeploymentService.credentialEndpointAllowed( + java.net.URI.create("https://control.example.test"), false)); + assertFalse(PluginDeploymentService.credentialEndpointAllowed( + java.net.URI.create("http://192.0.2.10:8080"), false)); + assertFalse(PluginDeploymentService.credentialEndpointAllowed( + java.net.URI.create("http://127.0.0.1:8080"), false)); + assertTrue(PluginDeploymentService.credentialEndpointAllowed( + java.net.URI.create("http://127.0.0.1:8080"), true)); + assertTrue(PluginDeploymentService.credentialEndpointAllowed( + java.net.URI.create("http://[::1]:8080"), true)); + assertFalse(PluginDeploymentService.credentialEndpointAllowed( + java.net.URI.create("http://127.example.com:8080"), true)); + } + + @Test void backendIgnoresMatchingMarkerOnlyWhenTargetIsValidThenRestagesWhenCorrupted() throws Exception { + byte[] artifact = jar("name: VotingPlugin\n"); + PluginDeploymentService service = PluginDeploymentService.backend(directory.resolve("update"), Path.of("VotingPlugin.jar")); + PluginDeploymentService.Task task = task(artifact); + + service.stage(task, new ByteArrayInputStream(artifact), () -> true); + assertArrayEquals(artifact, Files.readAllBytes(directory.resolve("update/VotingPlugin.jar"))); + + byte[] corrupted = new byte[artifact.length]; + Arrays.fill(corrupted, (byte) 0x42); + Files.write(directory.resolve("update/VotingPlugin.jar"), corrupted); + assertArrayEquals(corrupted, Files.readAllBytes(directory.resolve("update/VotingPlugin.jar"))); + + assertTrue(service.stage(task, new ByteArrayInputStream(artifact), () -> true).success()); + assertArrayEquals(artifact, Files.readAllBytes(directory.resolve("update/VotingPlugin.jar"))); + } + + @Test void invalidPluginIdentityAndDigestNeverReachTheUpdateFolder() throws Exception { + byte[] wrongPlugin = jar("name: NotVotingPlugin\n"); + PluginDeploymentService service = PluginDeploymentService.backend(directory.resolve("update"), Path.of("VotingPlugin.jar")); + PluginDeploymentService.Result invalid = service.stage(task(wrongPlugin), new ByteArrayInputStream(wrongPlugin), () -> true); + assertFalse(invalid.success()); + assertEquals("INVALID_ARTIFACT", invalid.code()); + assertFalse(Files.exists(directory.resolve("update/VotingPlugin.jar"))); + + byte[] good = jar("name: VotingPlugin\n"); + PluginDeploymentService.Task mismatch = new PluginDeploymentService.Task(UUID.randomUUID(), "vp.jar", + "0".repeat(64), good.length, UUID.randomUUID()); + assertEquals("HASH_MISMATCH", service.stage(mismatch, new ByteArrayInputStream(good), () -> true).code()); + } + + @Test void proxyKeepsTheCurrentJarAsABackupBeforeReplacement() throws Exception { + Path current = directory.resolve("VotingPlugin.jar"); + byte[] original = jar("name: VotingPlugin\nversion: 0\n"); + byte[] first = jar("name: VotingPlugin\nversion: 1\n"); + byte[] second = jar("name: VotingPlugin\nversion: 2\n"); + Files.write(current, original); + PluginDeploymentService service = PluginDeploymentService.proxy(current); + + assertEquals("RESTART_REQUIRED", service.stage(task(first), new ByteArrayInputStream(first), () -> true).code()); + assertEquals("RESTART_REQUIRED", service.stage(task(second), new ByteArrayInputStream(second), () -> true).code()); + assertArrayEquals(first, Files.readAllBytes(directory.resolve("VotingPlugin.jar.control-backup"))); + assertArrayEquals(second, Files.readAllBytes(current)); + } + + @Test void proxyMissingMarkerRequiresNormalVerifiedStaging() throws Exception { + Path current = directory.resolve("VotingPlugin.jar"); + Path backup = directory.resolve("VotingPlugin.jar.control-backup"); + Path marker = directory.resolve("VotingPlugin.jar.control-deployment"); + byte[] original = jar("name: VotingPlugin\nversion: original\n"); + byte[] candidate = jar("name: VotingPlugin\nversion: candidate\n"); + PluginDeploymentService.Task task = task(candidate); + Files.write(current, candidate); + Files.write(backup, original); + PluginDeploymentService service = PluginDeploymentService.proxy(current); + + assertEquals("SIZE_MISMATCH", + service.stage(task, new ByteArrayInputStream(new byte[0]), () -> true).code()); + assertFalse(Files.exists(marker)); + assertArrayEquals(original, Files.readAllBytes(backup)); + assertEquals("RESTART_REQUIRED", + service.stage(task, new ByteArrayInputStream(candidate), () -> true).code()); + + assertArrayEquals(candidate, Files.readAllBytes(current)); + assertArrayEquals(candidate, Files.readAllBytes(backup)); + String state = Files.readString(marker, StandardCharsets.US_ASCII); + assertTrue(state.contains(task.deploymentId().toString())); + assertTrue(state.contains(task.sha256())); + assertTrue(state.contains(Long.toString(task.size()))); + } + + @Test void proxyDoesNotAcknowledgeAStagedArtifactWithoutItsBackup() throws Exception { + Path current = directory.resolve("VotingPlugin.jar"); + Path backup = directory.resolve("VotingPlugin.jar.control-backup"); + byte[] original = jar("name: VotingPlugin\nversion: original\n"); + byte[] candidate = jar("name: VotingPlugin\nversion: candidate\n"); + Files.write(current, original); + PluginDeploymentService service = PluginDeploymentService.proxy(current); + PluginDeploymentService.Task task = task(candidate); + assertEquals("RESTART_REQUIRED", service.stage(task, new ByteArrayInputStream(candidate), () -> true).code()); + Files.delete(backup); + assertEquals("SIZE_MISMATCH", service.stage(task, new ByteArrayInputStream(new byte[0]), () -> true).code()); + assertEquals("RESTART_REQUIRED", service.stage(task, new ByteArrayInputStream(candidate), () -> true).code()); + assertTrue(Files.isRegularFile(backup)); + } + + @Test void cancellationDuringCopyDoesNotPublish() throws Exception { + byte[] artifact = jar("name: VotingPlugin\n"); + PluginDeploymentService service = PluginDeploymentService.backend(directory.resolve("update"), Path.of("VotingPlugin.jar")); + PluginDeploymentService.Task task = task(artifact); + + assertEquals("RESTART_REQUIRED", service.stage(task, new ByteArrayInputStream(artifact), () -> true).code()); + String markerBefore = Files.readString(directory.resolve("update/VotingPlugin.jar.control-deployment")); + byte[] targetBefore = Files.readAllBytes(directory.resolve("update/VotingPlugin.jar")); + + AtomicBoolean active = new AtomicBoolean(true); + byte[] replacementArtifact = jar("name: VotingPlugin\nversion: replacement\n"); + InputStream body = new CancellingInputStream(new ByteArrayInputStream(replacementArtifact), active); + PluginDeploymentService.Task replacement = task(replacementArtifact); + assertEquals("CANCELLED", service.stage(replacement, body, active::get).code()); + + assertEquals(markerBefore, Files.readString(directory.resolve("update/VotingPlugin.jar.control-deployment"))); + assertArrayEquals(targetBefore, Files.readAllBytes(directory.resolve("update/VotingPlugin.jar"))); + assertFalse(active.get()); + } + + @Test void markerPublicationFailureRestoresThePreviousArtifact() throws Exception { + Path update = directory.resolve("update"); + Files.createDirectories(update); + byte[] original = jar("name: VotingPlugin\nversion: original\n"); + byte[] candidate = jar("name: VotingPlugin\nversion: candidate\n"); + Path target = update.resolve("VotingPlugin.jar"); + Files.write(target, original); + Path marker = update.resolve("VotingPlugin.jar.control-deployment"); + Files.createDirectory(marker); + Files.writeString(marker.resolve("keep"), "marker publication must fail"); + PluginDeploymentService service = PluginDeploymentService.backend(update, Path.of("VotingPlugin.jar")); + + assertThrows(IOException.class, () -> service.stage(task(candidate), new ByteArrayInputStream(candidate), () -> true)); + + assertArrayEquals(original, Files.readAllBytes(target)); + } + + @Test void proxyMarkerPublicationFailureRestoresTheCurrentJarAndKeepsItsBackup() throws Exception { + byte[] original = jar("name: VotingPlugin\nversion: original\n"); + byte[] candidate = jar("name: VotingPlugin\nversion: candidate\n"); + Path target = directory.resolve("VotingPlugin.jar"); + Files.write(target, original); + Path marker = directory.resolve("VotingPlugin.jar.control-deployment"); + Files.createDirectory(marker); + Files.writeString(marker.resolve("keep"), "marker publication must fail"); + PluginDeploymentService service = PluginDeploymentService.proxy(target); + + assertThrows(IOException.class, () -> service.stage(task(candidate), new ByteArrayInputStream(candidate), () -> true)); + + assertArrayEquals(original, Files.readAllBytes(target)); + assertArrayEquals(original, Files.readAllBytes(directory.resolve("VotingPlugin.jar.control-backup"))); + } + + @Test void cancellationClosesAStalledArtifactStream() throws Exception { + PluginDeploymentService service = PluginDeploymentService.backend(directory.resolve("update"), Path.of("VotingPlugin.jar")); + AtomicBoolean active = new AtomicBoolean(true); + BlockingInputStream stalled = new BlockingInputStream(active); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future result = executor.submit(() -> service.stage( + task(new byte[] { 1 }), stalled, active::get)); + assertTrue(stalled.awaitRead()); + service.cancel(); + assertEquals("CANCELLED", result.get(5, TimeUnit.SECONDS).code()); + assertFalse(Files.exists(directory.resolve("update/VotingPlugin.jar"))); + } finally { + executor.shutdownNow(); + } + } + + private static PluginDeploymentService.Task task(byte[] artifact) throws Exception { + String digest = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(artifact)); + return new PluginDeploymentService.Task(UUID.randomUUID(), "votingplugin.jar", digest, artifact.length, + UUID.randomUUID()); + } + + private static byte[] jar(String pluginYml) throws Exception { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (JarOutputStream jar = new JarOutputStream(bytes)) { + jar.putNextEntry(new JarEntry("plugin.yml")); + jar.write(pluginYml.getBytes(StandardCharsets.UTF_8)); + jar.closeEntry(); + jar.putNextEntry(new JarEntry("example/Main.class")); + jar.write(new byte[] { 0, 1, 2 }); + jar.closeEntry(); + } + return bytes.toByteArray(); + } + + private static class CancellingInputStream extends InputStream { + private final InputStream delegate; + private final AtomicBoolean active; + private boolean cancelled; + + private CancellingInputStream(InputStream delegate, AtomicBoolean active) { + this.delegate = delegate; + this.active = active; + } + + @Override + public int read(byte[] bytes, int start, int length) throws IOException { + int read = delegate.read(bytes, start, length); + if (!cancelled && read > 0) { + cancelled = true; + active.set(false); + } + return read; + } + + @Override + public int read() throws IOException { return delegate.read(); } + } + + private static class BlockingInputStream extends InputStream { + private final AtomicBoolean active; + private final CountDownLatch reading = new CountDownLatch(1); + private boolean closed; + + private BlockingInputStream(AtomicBoolean active) { this.active = active; } + + private boolean awaitRead() throws InterruptedException { return reading.await(5, TimeUnit.SECONDS); } + + @Override + public synchronized int read(byte[] bytes, int start, int length) throws IOException { + reading.countDown(); + while (!closed) { + try { wait(); } + catch (InterruptedException failure) { Thread.currentThread().interrupt(); throw new IOException(failure); } + } + throw new IOException("stream closed"); + } + + @Override + public int read() throws IOException { return read(new byte[1], 0, 1); } + + @Override + public synchronized void close() { + closed = true; + active.set(false); + notifyAll(); + } + } +} diff --git a/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/control/ControlConnectorTest.java b/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/control/ControlConnectorTest.java index caffbbc47..315447dfe 100644 --- a/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/control/ControlConnectorTest.java +++ b/VotingPlugin/src/test/java/com/bencodez/votingplugin/proxy/control/ControlConnectorTest.java @@ -1,5 +1,6 @@ package com.bencodez.votingplugin.proxy.control; +import com.bencodez.votingplugin.control.PluginDeploymentService; import com.bencodez.votingplugin.proxy.control.ControlConnector.ObservedBackend; import com.bencodez.votingplugin.proxy.control.ControlConnector.Request; import com.bencodez.votingplugin.proxy.control.ControlConnector.Response; @@ -12,6 +13,7 @@ import java.lang.reflect.Constructor; import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.net.http.HttpClient; import java.net.URI; import java.nio.file.Files; import java.nio.file.Path; @@ -23,6 +25,7 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -48,6 +51,17 @@ class ControlConnectorTest { @Test void responseBudgetCanCarryTheLargestEscapedManagedFileTask() { assertTrue(ControlConnector.MAX_RESPONSE_BYTES >= ProxyConfigurationFileService.MAX_BYTES * 6); } + + @Test void deploymentTaskRejectsUnknownV1Fields() { + JsonObject task = new JsonObject(); + task.addProperty("deploymentId", "00000000-0000-0000-0000-000000000001"); + task.addProperty("artifactId", "VotingPlugin.jar"); + task.addProperty("sha256", "a".repeat(64)); + task.addProperty("size", "1"); + task.addProperty("attemptId", "00000000-0000-0000-0000-000000000002"); + task.addProperty("unexpected", "value"); + assertThrows(RuntimeException.class, () -> ControlConnector.deploymentTask(task)); + } private ControlConnector connector; @BeforeEach void setUp() { @@ -88,6 +102,18 @@ class ControlConnectorTest { assertEquals(1, secondSnapshot.get("sequence").getAsLong()); } + @Test void deploymentCapabilityIsAdvertisedOnlyWhenProxyStagingIsReady() { + JsonObject unavailable = new JsonObject(); + ControlConnector.addCapabilities(unavailable, true, true, true, true, false); + assertFalse(unavailable.getAsJsonArray("capabilities").asList().stream() + .anyMatch(value -> "plugin.deploy.v1".equals(value.getAsString()))); + + JsonObject ready = new JsonObject(); + ControlConnector.addCapabilities(ready, true, true, true, true, true); + assertTrue(ready.getAsJsonArray("capabilities").asList().stream() + .anyMatch(value -> "plugin.deploy.v1".equals(value.getAsString()))); + } + @Test void unavailableAuthenticationProtocolAndMalformedResponsesOnlyChangeConnectorState() { transport.nextPrimary = new Response(401, "{\"error\":{}}"); connector.cycle(); @@ -879,6 +905,51 @@ class ControlConnectorTest { assertEquals(Status.STOPPED, connector.status()); } + @Test void closeCancelsARealQueuedDeploymentBeforeAwaitingItsOperation() throws Exception { + connector.close(); + Path update = dataDirectory.resolve("update"); + connector = deploymentConnector(PluginDeploymentService.backend(update, dataDirectory.resolve("VotingPlugin.jar"))); + setConnectorField("registered", true); + setConnectorField("deploymentsAccepted", true); + setConnectorField("status", Status.CONNECTED); + Field executorField = ControlConnector.class.getDeclaredField("deploymentExecutor"); + executorField.setAccessible(true); + ExecutorService deploymentExecutor = (ExecutorService) executorField.get(connector); + CountDownLatch workerStarted = new CountDownLatch(1); + deploymentExecutor.submit(() -> { + workerStarted.countDown(); + try { + new CountDownLatch(1).await(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } + }); + assertTrue(workerStarted.await(2, TimeUnit.SECONDS)); + UUID deploymentId = UUID.randomUUID(); + transport.nextPrimary = new Response(200, "{\"deploymentId\":\"" + deploymentId + + "\",\"artifactId\":\"VotingPlugin.jar\",\"sha256\":\"" + "a".repeat(64) + + "\",\"size\":\"1\",\"attemptId\":\"" + UUID.randomUUID() + "\"}"); + Method poll = ControlConnector.class.getDeclaredMethod("pollDeployments"); + poll.setAccessible(true); + poll.invoke(connector); + Field operationField = ControlConnector.class.getDeclaredField("activeOperation"); + operationField.setAccessible(true); + CompletableFuture operation = (CompletableFuture) operationField.get(connector); + assertNotNull(operation, "the actual deployment polling path must own the queued operation"); + + connector.close(); + + assertTrue(operation.isDone(), "shutdown must complete a deployment discarded before its executor starts it"); + assertFalse(Files.exists(update.resolve("VotingPlugin.jar")), "queued deployment work must not stage an artifact"); + } + + @Test void proxyDeploymentIsNotAdvertisedOnWindowsWhereTheLiveJarCannotBeReplaced() { + assertFalse(ControlConnector.proxyDeploymentSupported("Windows 11")); + assertFalse(ControlConnector.proxyDeploymentSupported("Windows Server 2022")); + assertTrue(ControlConnector.proxyDeploymentSupported("Linux")); + assertTrue(ControlConnector.proxyDeploymentSupported("Darwin")); + } + @Test void backoffIsBoundedExponentialAndJitteredWithoutSleeping() { assertEquals(1000, ControlConnector.backoffMillis(1, 0)); assertEquals(2000, ControlConnector.backoffMillis(2, 0)); @@ -909,6 +980,25 @@ private Settings settings() { URI.create("http://127.0.0.1:8080"), 30, 3000, 5000); } + private ControlConnector deploymentConnector(PluginDeploymentService deployments) throws Exception { + Constructor constructor = ControlConnector.class.getDeclaredConstructor(Settings.class, + ScheduledExecutorService.class, Transport.class, Supplier.class, Consumer.class, UUID.class, + LongSupplier.class, ProxyRoutingConfigurationService.class, Path.class, ProxyControlResultStore.Route.class, + boolean.class, Runnable.class, Function.class, ProxyMethodConfigurationService.class, Runnable.class, + ProxyConfigurationFileService.class, PluginDeploymentService.class, HttpClient.class, String.class, + boolean.class); + constructor.setAccessible(true); + return constructor.newInstance(settings(), scheduler, transport, (Supplier>) List::of, + (Consumer) logs::add, UUID.randomUUID(), (LongSupplier) () -> 0L, null, null, null, false, + null, null, null, null, null, deployments, HttpClient.newHttpClient(), "credential", false); + } + + private void setConnectorField(String name, Object value) throws Exception { + Field field = ControlConnector.class.getDeclaredField(name); + field.setAccessible(true); + field.set(connector, value); + } + private JsonObject submittedResult() { return JsonParser.parseString(transport.requests.stream().filter(request -> request.path().endsWith("/result")) .findFirst().orElseThrow().body()).getAsJsonObject(); diff --git a/docs/control-agent-contract.md b/docs/control-agent-contract.md index 60468f37c..dc1e5d6ad 100644 --- a/docs/control-agent-contract.md +++ b/docs/control-agent-contract.md @@ -143,6 +143,65 @@ redacted-editor task; turning `useMainMySQL` off alone does not invent credentia unknown fields. `READ` returns current typed state and its revision, `PREVIEW` computes changes and produces an approval, and only the approved `APPLY` writes atomically, reloads, and rolls back on reload failure, like every other quick setup. +## Verified plugin staging (`plugin.deploy.v1`) + +This optional capability stages one verified VotingPlugin JAR for the node's next process restart. It is never a hot reload +and never restarts a proxy or backend automatically. A node advertises it only when all of the following are true: + +- the connector is the currently enabled Control route, not a recovery-only connector draining an older durable result; +- a safe local staging target was prepared; +- the Control endpoint is HTTPS, or it is the already-proven direct same-node hosted HTTP listener. + +Arbitrary private-network HTTP does not qualify for deployment because the artifact request carries the node bearer +credential. The shared staging service enforces the same transport rule again before sending that credential. + +Control leases deployment work separately from configuration operations: + +```http +POST /api/v1/nodes/{nodeId}/deployments +Content-Type: application/json + +{"sessionId":""} +``` + +A `204` means no work. A `200` task contains `deploymentId`, `artifactId`, lowercase SHA-256 `sha256`, byte +`size`, and `attemptId`. Artifact size is limited to 64 MiB. The node downloads the exact leased artifact with: + +```http +GET /api/v1/nodes/{nodeId}/deployments/{deploymentId}/artifact +Authorization: Bearer +X-Node-Session: +X-Deployment-Attempt: +``` + +The node independently verifies the exact size and SHA-256, bounded ZIP/JAR structure, root `plugin.yml`, and +`name: VotingPlugin` before publication. Bukkit nodes stage to the server update folder; proxy nodes atomically replace +their discovered plugin JAR only after creating a durable `.control-backup`. A small durable +`.control-deployment` marker is published before the verified target is moved into place, so a lost result +acknowledgement does not apply the same artifact twice. On Bukkit, a consumed staged update must match the installed JAR. +On proxies, both a safe backup and a target JAR matching the marker's digest are required for acknowledgement; a missing +marker requires a fresh verified download and staging attempt. + +The result is posted to: + +```http +POST /api/v1/nodes/{nodeId}/deployments/{deploymentId}/result +Content-Type: application/json + +{ + "sessionId":"", + "attemptId":"", + "success":true, + "code":"RESTART_REQUIRED", + "message":"Plugin update staged; restart is required" +} +``` + +Failure codes include `CANCELLED`, `DOWNLOAD_FAILED`, `SIZE_MISMATCH`, `HASH_MISMATCH`, +`INVALID_ARTIFACT`, `STAGING_FAILED`, `DEPLOYMENT_FAILED`, and `INSECURE_ENDPOINT`. Deployment polling is +serialized with the connector lifecycle, uses the normal bounded failure backoff, and is cancelled during shutdown before +transport teardown. + ## Inspection transport An enrolled Bukkit node advertises `data.inspect.v1`. Once Control includes it in `acceptedCapabilities`, the node polls: diff --git a/docs/control-connector.md b/docs/control-connector.md index ae40c78b3..760218ffc 100644 --- a/docs/control-connector.md +++ b/docs/control-connector.md @@ -201,6 +201,29 @@ with the new settings. This prevents the old and new children from racing for th the currently running Control child unchanged. If the backend restarts with a pending result, the durable journal also restores the previous hosted settings until that result is acknowledged, preserving the recovery connector's endpoint. +## Verified VotingPlugin updates + +Nodes that negotiate `plugin.deploy.v1` can stage a VotingPlugin JAR supplied by Control for the next normal process +restart. This capability is deliberately separate from configuration control and from hosted-Control self-updates. +VotingPlugin never hot-reloads itself and never restarts the server or proxy automatically. + +Deployment is available only on the currently enabled Control route. Recovery-only connectors that exist solely to +acknowledge an older durable result never advertise or poll this capability. The node also requires a credential-safe +artifact transport: HTTPS is accepted generally; HTTP is accepted only when the existing hosted-Control checks prove the +endpoint is the direct same-node listener. A LAN/private HTTP endpoint may still be used for ordinary Control operations, +but it is intentionally ineligible for credentialed plugin-JAR staging. + +Control leases deployment work through `POST /api/v1/nodes/{nodeId}/deployments`. The node downloads the artifact through +the matching deployment artifact endpoint with its bearer credential plus exact session and attempt headers, then +independently verifies the 64 MiB size bound, SHA-256, JAR structure, and root `plugin.yml` identity. Bukkit stages the +verified JAR in the configured update folder; BungeeCord/Velocity retain a durable backup before atomically replacing the +running plugin JAR on disk. A durable deployment marker makes lost result acknowledgements idempotent, including the +post-restart Bukkit state where the server has already consumed the staged update JAR. + +Successful staging reports `RESTART_REQUIRED`. The administrator chooses when to restart. See +[the Control agent contract](control-agent-contract.md#verified-plugin-staging-plugindeployv1) for the exact task, download, +result, transport, and failure-code contract. + ## Discovery semantics Both platforms use the same connector implementation and protocol version `1`. Each proxy process creates a new session