diff --git a/api/src/main/java/com/velocitypowered/api/event/connection/ConnectionEstablishEvent.java b/api/src/main/java/com/velocitypowered/api/event/connection/ConnectionEstablishEvent.java
new file mode 100644
index 0000000000..adb8f4c86c
--- /dev/null
+++ b/api/src/main/java/com/velocitypowered/api/event/connection/ConnectionEstablishEvent.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright (C) 2018-2021 Velocity Contributors
+ *
+ * The Velocity API is licensed under the terms of the MIT License. For more details,
+ * reference the LICENSE file in the api top-level directory.
+ */
+
+package com.velocitypowered.api.event.connection;
+
+import com.google.common.base.Preconditions;
+import com.velocitypowered.api.event.ResultedEvent;
+import com.velocitypowered.api.event.annotation.AwaitingEvent;
+import com.velocitypowered.api.network.HandshakeIntent;
+import com.velocitypowered.api.proxy.InboundConnection;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Event called when a connection is initially established with the proxy.
+ * Velocity will wait for this event to fire before continuing with the connection.
+ */
+@AwaitingEvent
+public class ConnectionEstablishEvent implements ResultedEvent<ResultedEvent.GenericResult> {
+  private final InboundConnection connection;
+  private final HandshakeIntent intention;
+  private GenericResult result = GenericResult.allowed();
+
+  public ConnectionEstablishEvent(
+          final @NonNull InboundConnection connection,
+          final @Nullable HandshakeIntent intention
+  ) {
+    this.connection = Preconditions.checkNotNull(connection, "connection");
+    this.intention = intention;
+  }
+
+  /**
+   * Returns the inbound connection that is being established.
+   *
+   * @return the connection
+   */
+  public @NonNull InboundConnection getConnection() {
+    return this.connection;
+  }
+
+  /**
+   * Returns the intention for which the connection is being established.
+   *
+   * @return the intention
+   */
+  public HandshakeIntent getIntention() {
+    return this.intention;
+  }
+
+  @Override
+  public GenericResult getResult() {
+    return this.result;
+  }
+
+  @Override
+  public void setResult(final @NonNull GenericResult result) {
+    this.result = Preconditions.checkNotNull(result, "result");
+  }
+}
diff --git a/proxy/src/main/java/com/velocitypowered/proxy/connection/client/HandshakeSessionHandler.java b/proxy/src/main/java/com/velocitypowered/proxy/connection/client/HandshakeSessionHandler.java
index 80260d2b47..e8a33d681d 100644
--- a/proxy/src/main/java/com/velocitypowered/proxy/connection/client/HandshakeSessionHandler.java
+++ b/proxy/src/main/java/com/velocitypowered/proxy/connection/client/HandshakeSessionHandler.java
@@ -19,6 +19,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.velocitypowered.api.event.connection.ConnectionEstablishEvent;
 import com.velocitypowered.api.event.connection.ConnectionHandshakeEvent;
 import com.velocitypowered.api.network.HandshakeIntent;
 import com.velocitypowered.api.network.ProtocolState;
@@ -94,22 +95,35 @@ public boolean handle(final HandshakePacket handshake) {
     } else {
       final InitialInboundConnection ic = new InitialInboundConnection(connection,
               cleanVhost(handshake.getServerAddress()), handshake);
-      if (handshake.getIntent() == HandshakeIntent.TRANSFER
-              && !server.getConfiguration().isAcceptTransfers()) {
-        ic.disconnect(Component.translatable("multiplayer.disconnect.transfers_disabled"));
-        return true;
-      }
-      connection.setProtocolVersion(handshake.getProtocolVersion());
-      connection.setAssociation(ic);
-
-      switch (nextState) {
-        case STATUS -> connection.setActiveSessionHandler(StateRegistry.STATUS,
-              new StatusSessionHandler(server, ic));
-        case LOGIN -> this.handleLogin(handshake, ic);
-        default ->
-          // If you get this, it's a bug in Velocity.
-          throw new AssertionError("getStateForProtocol provided invalid state!");
-      }
+      // Handle connection establish event.
+      connection.setAutoReading(false);
+      server.getEventManager()
+          .fire(new ConnectionEstablishEvent(ic, handshake.getIntent()))
+          .thenAcceptAsync(result -> {
+            // Clean up the disabling of auto-read.
+            connection.setAutoReading(true);
+
+            if (!result.getResult().isAllowed()) {
+              connection.close(true);
+            } else {
+              if (handshake.getIntent() == HandshakeIntent.TRANSFER
+                      && !server.getConfiguration().isAcceptTransfers()) {
+                ic.disconnect(Component.translatable("multiplayer.disconnect.transfers_disabled"));
+                return;
+              }
+              connection.setProtocolVersion(handshake.getProtocolVersion());
+              connection.setAssociation(ic);
+
+              switch (nextState) {
+                case STATUS -> connection.setActiveSessionHandler(StateRegistry.STATUS,
+                      new StatusSessionHandler(server, ic));
+                case LOGIN -> this.handleLogin(handshake, ic);
+                default ->
+                  // If you get this, it's a bug in Velocity.
+                  throw new AssertionError("getStateForProtocol provided invalid state!");
+              }
+            }
+          });
     }
 
     return true;