diff --git a/src/metrics_server/README.md b/src/metrics_server/README.md
index 45f03726..96824f49 100644
--- a/src/metrics_server/README.md
+++ b/src/metrics_server/README.md
@@ -35,6 +35,7 @@ The metrics server supports two URL paths:
         timestampUtcMs: number,
         dataLimit: {
             enabled: boolean
+            perKeyLimitCount: number
         }
     }
     ```
diff --git a/src/metrics_server/feature_metrics.spec.ts b/src/metrics_server/feature_metrics.spec.ts
index 7970887f..4f766221 100644
--- a/src/metrics_server/feature_metrics.spec.ts
+++ b/src/metrics_server/feature_metrics.spec.ts
@@ -54,6 +54,24 @@ describe('isValidFeatureMetricsReport', () => {
     };
     expect(isValidFeatureMetricsReport(report)).toBeTruthy();
   });
+  it('returns true for valid report with per-key data limit count', () => {
+    const report = {
+      serverId: 'id',
+      serverVersion: '0.0.0',
+      timestampUtcMs: 123456,
+      dataLimit: {enabled: true, perKeyLimitCount: 1}
+    };
+    expect(isValidFeatureMetricsReport(report)).toBeTruthy();
+  });
+  it('returns false for report with negative per-key data limit count', () => {
+    const report = {
+      serverId: 'id',
+      serverVersion: '0.0.0',
+      timestampUtcMs: 123456,
+      dataLimit: {enabled: true, perKeyLimitCount: -1}
+    };
+    expect(isValidFeatureMetricsReport(report)).toBeFalsy();
+  });
   it('returns false for missing report', () => {
     expect(isValidFeatureMetricsReport(undefined)).toBeFalsy();
   });
diff --git a/src/metrics_server/feature_metrics.ts b/src/metrics_server/feature_metrics.ts
index 484e3de9..2d3ce2fb 100644
--- a/src/metrics_server/feature_metrics.ts
+++ b/src/metrics_server/feature_metrics.ts
@@ -66,10 +66,18 @@ export function isValidFeatureMetricsReport(obj: any): obj is DailyFeatureMetric
     return false;
   }
 
-  // Validate individual feature records.
+  // Validate the server data limit feature
   if (typeof obj.dataLimit.enabled !== 'boolean') {
     return false;
   }
 
-  return true;
+  // Validate the per-key data limit feature
+  const perKeyLimitCount = obj.dataLimit.perKeyLimitCount;
+  if(perKeyLimitCount === undefined) {
+    return true;
+  }
+  if (typeof perKeyLimitCount === 'number') {
+    return obj.dataLimit.perKeyLimitCount >= 0;
+  }
+  return false;
 }
diff --git a/src/metrics_server/model.ts b/src/metrics_server/model.ts
index 6f7bccd6..e355e824 100644
--- a/src/metrics_server/model.ts
+++ b/src/metrics_server/model.ts
@@ -36,4 +36,5 @@ export interface DailyFeatureMetricsReport {
 
 export interface DailyDataLimitMetricsReport {
   enabled: boolean;
+  perKeyLimitCount?: number;
 }
diff --git a/src/metrics_server/test_integration_action.sh b/src/metrics_server/test_integration_action.sh
index 7a8f226a..03f6507d 100755
--- a/src/metrics_server/test_integration_action.sh
+++ b/src/metrics_server/test_integration_action.sh
@@ -41,6 +41,7 @@ USER_ID2=$(uuidgen)
 # BYTES_TRANSFERRED2 < BYTES_TRANSFERRED1 so we can order the records before comparing them.
 BYTES_TRANSFERRED1=$((2 + RANDOM % 100))
 BYTES_TRANSFERRED2=$(($BYTES_TRANSFERRED1 - 1))
+PER_KEY_LIMIT_COUNT=$((RANDOM))
 
 echo "Using tmp directory $TMPDIR"
 
@@ -67,7 +68,8 @@ cat << EOF > $FEATURES_REQUEST
   "serverVersion": "$SERVER_VERSION",
   "timestampUtcMs": $TIMESTAMP,
   "dataLimit": {
-    "enabled": false
+    "enabled": false,
+    "perKeyLimitCount": $PER_KEY_LIMIT_COUNT
   }
 }
 EOF
@@ -78,7 +80,7 @@ cat << EOF > $CONNECTIONS_EXPECTED_RESPONSE
 [{"bytesTransferred":"$BYTES_TRANSFERRED1","countries":["US","NL"],"serverId":"$SERVER_ID","userId":"$USER_ID1"},{"bytesTransferred":"$BYTES_TRANSFERRED2","countries":["UK"],"serverId":"$SERVER_ID","userId":"$USER_ID2"}]
 EOF
 cat << EOF > $FEATURES_EXPECTED_RESPONSE
-[{"dataLimit":{"enabled":"false"},"serverId":"$SERVER_ID","serverVersion":"$SERVER_VERSION"}]
+[{"dataLimit":{"enabled":"false","perKeyLimitCount":"$PER_KEY_LIMIT_COUNT"},"serverId":"$SERVER_ID","serverVersion":"$SERVER_VERSION"}]
 EOF
 
 echo "Connections request:"
diff --git a/src/shadowbox/integration_test/test.sh b/src/shadowbox/integration_test/test.sh
index 735caec4..6706ab93 100755
--- a/src/shadowbox/integration_test/test.sh
+++ b/src/shadowbox/integration_test/test.sh
@@ -53,12 +53,16 @@ function wait_for_resource() {
   until curl --silent --insecure $URL > /dev/null; do sleep 1; done
 }
 
+function util_jq() {
+  docker exec -i $UTIL_CONTAINER jq "$@"
+}
+
 # Takes the JSON from a /access-keys POST request and returns the appropriate
 # ss-local arguments to connect to that user/instance.
 function ss_arguments_for_user() {
-  declare -r SS_INSTANCE_CIPHER=$(echo $1 | docker exec -i $UTIL_CONTAINER jq -r .method)
-  declare -r SS_INSTANCE_PASSWORD=$(echo $1 | docker exec -i $UTIL_CONTAINER jq -r .password)
-  declare -r SS_INSTANCE_PORT=$(echo $1 | docker exec -i $UTIL_CONTAINER jq .port)
+  declare -r SS_INSTANCE_CIPHER=$(echo $1 | util_jq -r .method)
+  declare -r SS_INSTANCE_PASSWORD=$(echo $1 | util_jq -r .password)
+  declare -r SS_INSTANCE_PORT=$(echo $1 | util_jq .port)
   echo -cipher "$SS_INSTANCE_CIPHER" -password "$SS_INSTANCE_PASSWORD" -c "shadowbox:$SS_INSTANCE_PORT"
 }
 
@@ -136,44 +140,89 @@ function cleanup() {
     sleep 0.1
   done
 
-  # Verify the server blocks requests to hosts on private addresses.
-  # Exit code 52 is "Empty server response".
-  client_curl -x socks5h://localhost:$LOCAL_SOCKS_PORT $TARGET_IP &> /dev/null \
-    && fail "Target host in a private network accessible through shadowbox" || (($? == 52))
+  function test_networking() {
+    # Verify the server blocks requests to hosts on private addresses.
+    # Exit code 52 is "Empty server response".
+    client_curl -x socks5h://localhost:$LOCAL_SOCKS_PORT $TARGET_IP &> /dev/null \
+      && fail "Target host in a private network accessible through shadowbox" || (($? == 52))
 
-  # Verify we can retrieve the internet target URL.
-  client_curl -x socks5h://localhost:$LOCAL_SOCKS_PORT $INTERNET_TARGET_URL \
-    || fail "Could not fetch $INTERNET_TARGET_URL through shadowbox."
+    # Verify we can retrieve the internet target URL.
+    client_curl -x socks5h://localhost:$LOCAL_SOCKS_PORT $INTERNET_TARGET_URL \
+      || fail "Could not fetch $INTERNET_TARGET_URL through shadowbox."
 
-  # Verify we can't access the URL anymore after the key is deleted
-  client_curl --insecure -X DELETE ${SB_API_URL}/access-keys/0 > /dev/null
-  # Exit code 56 is "Connection reset by peer".
-  client_curl -x socks5h://localhost:$LOCAL_SOCKS_PORT $INTERNET_TARGET_URL &> /dev/null \
-    && fail "Deleted access key is still active" || (($? == 56))
+    # Verify we can't access the URL anymore after the key is deleted
+    client_curl --insecure -X DELETE ${SB_API_URL}/access-keys/0 > /dev/null
+    # Exit code 56 is "Connection reset by peer".
+    client_curl -x socks5h://localhost:$LOCAL_SOCKS_PORT $INTERNET_TARGET_URL &> /dev/null \
+      && fail "Deleted access key is still active" || (($? == 56))
+  }
 
-  # Verify that we can change the port for new access keys
-  client_curl --insecure -X PUT -H "Content-Type: application/json" -d '{"port": 12345}' ${SB_API_URL}/server/port-for-new-access-keys \
-    || fail "Couldn't change the port for new access keys"
+  function test_port_for_new_keys() {
+    # Verify that we can change the port for new access keys
+    client_curl --insecure -X PUT -H "Content-Type: application/json" -d '{"port": 12345}' ${SB_API_URL}/server/port-for-new-access-keys \
+      || fail "Couldn't change the port for new access keys"
 
-  ACCESS_KEY_JSON=$(client_curl --insecure -X POST ${SB_API_URL}/access-keys \
-    || fail "Couldn't get a new access key after changing port")
+    ACCESS_KEY_JSON=$(client_curl --insecure -X POST ${SB_API_URL}/access-keys \
+      || fail "Couldn't get a new access key after changing port")
+    
+    if [[ "${ACCESS_KEY_JSON}" != *'"port":12345'* ]]; then
+      fail "Port for new access keys wasn't changed.  Newly created access key: ${ACCESS_KEY_JSON}"
+    fi
+  }
+
+  function test_hostname_for_new_keys() { 
+    # Verify that we can change the hostname for new access keys
+    NEW_HOSTNAME="newhostname"
+    client_curl --insecure -X PUT -H 'Content-Type: application/json' -d '{"hostname": "'${NEW_HOSTNAME}'"}' ${SB_API_URL}/server/hostname-for-access-keys \
+      || fail "Couldn't change hostname for new access keys"
+
+    ACCESS_KEY_JSON=$(client_curl --insecure -X POST ${SB_API_URL}/access-keys \
+      || fail "Couldn't get a new access key after changing hostname")
+    
+    if [[ "${ACCESS_KEY_JSON}" != *"@${NEW_HOSTNAME}:"* ]]; then
+      fail "Hostname for new access keys wasn't changed.  Newly created access key: ${ACCESS_KEY_JSON}"
+    fi
+  }
+
+  function test_default_data_limit() {
+    # Verify that we can create default data limits
+    client_curl --insecure -X PUT -H 'Content-Type: application/json' -d '{"limit": {"bytes": 1000}}' \
+        ${SB_API_URL}/server/access-key-data-limit \
+      || fail "Couldn't create default data limit"
+    client_curl --insecure ${SB_API_URL}/server | grep -q 'accessKeyDataLimit' || fail 'Default data limit not set'
+   
+   # Verify that we can remove default data limits
+    client_curl --insecure -X DELETE ${SB_API_URL}/server/access-key-data-limit \
+      || fail "Couldn't remove default data limit"
+    client_curl --insecure ${SB_API_URL}/server | grep -vq 'accessKeyDataLimit' || fail 'Default data limit not removed'
+  }
+
+  function test_per_key_data_limits() {
+    # Verify that we can create per-key data limits
+    ACCESS_KEY_ID=$(client_curl --insecure -X POST ${SB_API_URL}/access-keys | util_jq -re .id \
+      || fail "Couldn't get a key to test custom data limits")
+
+    client_curl --insecure -X PUT -H 'Content-Type: application/json' -d '{"limit": {"bytes": 1000}}' \
+        ${SB_API_URL}/access-keys/${ACCESS_KEY_ID}/data-limit \
+      || fail "Couldn't create per-key data limit"
+    client_curl --insecure ${SB_API_URL}/access-keys \
+      | util_jq -e ".accessKeys[] | select(.id == \"${ACCESS_KEY_ID}\") | .dataLimit.bytes" \
+      || fail 'Per-key data limit not set'
+   
+    # Verify that we can remove per-key data limits
+    client_curl --insecure -X DELETE ${SB_API_URL}/access-keys/${ACCESS_KEY_ID}/data-limit \
+      || fail "Couldn't remove per-key data limit"
+    ! client_curl --insecure ${SB_API_URL}/access-keys \
+      | util_jq -e ".accessKeys[] | select(.id == \"${ACCESS_KEY_ID}\") | .dataLimit.bytes" \
+      || fail 'Per-key data limit not removed'
+  }
   
-  if [[ "${ACCESS_KEY_JSON}" != *'"port":12345'* ]]; then
-    fail "Port for new access keys wasn't changed.  Newly created access key: ${ACCESS_KEY_JSON}"
-  fi
+  test_networking
+  test_port_for_new_keys 
+  test_hostname_for_new_keys
+  test_default_data_limit
+  test_per_key_data_limits
 
-  # Verify that we can change the hostname for new access keys
-  NEW_HOSTNAME="newhostname"
-  client_curl --insecure -X PUT -H 'Content-Type: application/json' -d '{"hostname": "'${NEW_HOSTNAME}'"}' ${SB_API_URL}/server/hostname-for-access-keys \
-    || fail "Couldn't change hostname for new access keys"
-
-  ACCESS_KEY_JSON=$(client_curl --insecure -X POST ${SB_API_URL}/access-keys \
-    || fail "Couldn't get a new access key after changing hostname")
-  
-  if [[ "${ACCESS_KEY_JSON}" != *"@${NEW_HOSTNAME}:"* ]]; then
-    fail "Hostname for new access keys wasn't changed.  Newly created access key: ${ACCESS_KEY_JSON}"
-  fi
-  
   # Verify no errors occurred.
   readonly SHADOWBOX_LOG=$OUTPUT_DIR/shadowbox-log.txt
   if docker logs $SHADOWBOX_CONTAINER 2>&1 | tee $SHADOWBOX_LOG | egrep -q "^E|level=error|ERROR:"; then
diff --git a/src/shadowbox/model/access_key.ts b/src/shadowbox/model/access_key.ts
index 46be28a2..00dfd097 100644
--- a/src/shadowbox/model/access_key.ts
+++ b/src/shadowbox/model/access_key.ts
@@ -44,6 +44,8 @@ export interface AccessKey {
   readonly proxyParams: ProxyParams;
   // Whether the access key has exceeded the data transfer limit.
   readonly isOverDataLimit: boolean;
+  // The key's current data limit.  If it exists, it overrides the server default data limit.
+  readonly dataLimit?: DataLimit;
 }
 
 export interface AccessKeyRepository {
@@ -62,7 +64,11 @@ export interface AccessKeyRepository {
   // Gets the metrics id for a given Access Key.
   getMetricsId(id: AccessKeyId): AccessKeyMetricsId|undefined;
   // Sets a data transfer limit for all access keys.
-  setAccessKeyDataLimit(limit: DataLimit): Promise<void>;
+  setDefaultDataLimit(limit: DataLimit): void;
   // Removes the access key data transfer limit.
-  removeAccessKeyDataLimit(): Promise<void>;
+  removeDefaultDataLimit(): void;
+  // Sets access key `id` to use the given custom data limit.
+  setAccessKeyDataLimit(id: AccessKeyId, limit: DataLimit): void;
+  // Removes the custom data limit from access key `id`.
+  removeAccessKeyDataLimit(id: AccessKeyId): void;
 }
diff --git a/src/shadowbox/model/errors.ts b/src/shadowbox/model/errors.ts
index bcfaf402..ddad5a28 100644
--- a/src/shadowbox/model/errors.ts
+++ b/src/shadowbox/model/errors.ts
@@ -38,9 +38,3 @@ export class AccessKeyNotFound extends OutlineError {
     super(`Access key "${accessKeyId}" not found`);
   }
 }
-
-export class InvalidAccessKeyDataLimit extends OutlineError {
-  constructor() {
-    super('Must provide a limit with a non-negative integer value for "bytes"');
-  }
-}
diff --git a/src/shadowbox/server/api.yml b/src/shadowbox/server/api.yml
index 88f07e62..75537ee9 100644
--- a/src/shadowbox/server/api.yml
+++ b/src/shadowbox/server/api.yml
@@ -204,6 +204,52 @@ paths:
           description: Access key renamed successfully
         '404':
           description: Access key inexistent
+  /access-keys/{id}/data-limit:
+    put:
+      description: Sets a data limit for the given access key
+      tags:
+        - Access Key
+        - Limit
+      parameters:
+        - name: id
+          in: path
+          required: true
+          description: The id of the access key
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/DataLimit"
+            examples:
+              '0':
+                value: "{limit: {bytes: 10000}}"
+      responses:
+        '204':
+          description: Access key limit set successfully
+        '400':
+          description: Invalid data limit
+        '404':
+          description: Access key inexistent
+    delete:
+      description: Removes the data limit on the given access key.
+      tags:
+        - Access Key
+        - Limit
+      parameters:
+        - name: id
+          in: path
+          required: true
+          description: The id of the access key
+          schema:
+            type: string
+      responses:
+        '204':
+          description: Access key limit deleted successfully.
+        '404':
+          description: Access key inexistent
   /metrics/transfer:
     get:
       description: Returns the data transferred per access key
diff --git a/src/shadowbox/server/main.ts b/src/shadowbox/server/main.ts
index 6dfcf9f4..bc573391 100644
--- a/src/shadowbox/server/main.ts
+++ b/src/shadowbox/server/main.ts
@@ -185,7 +185,7 @@ async function main() {
   const managerMetrics = new PrometheusManagerMetrics(prometheusClient);
   const metricsCollector = new RestMetricsCollectorClient(metricsCollectorUrl);
   const metricsPublisher: SharedMetricsPublisher = new OutlineSharedMetricsPublisher(
-      new RealClock(), serverConfig, metricsReader, toMetricsId, metricsCollector);
+      new RealClock(), serverConfig, accessKeyConfig, metricsReader, toMetricsId, metricsCollector);
   const managerService = new ShadowsocksManagerService(
       process.env.SB_DEFAULT_SERVER_NAME || 'Outline Server', serverConfig, accessKeyRepository,
       managerMetrics, metricsPublisher);
diff --git a/src/shadowbox/server/manager_service.spec.ts b/src/shadowbox/server/manager_service.spec.ts
index d0bc19cf..5cd04f68 100644
--- a/src/shadowbox/server/manager_service.spec.ts
+++ b/src/shadowbox/server/manager_service.spec.ts
@@ -32,7 +32,7 @@ interface ServerInfo {
 const NEW_PORT = 12345;
 const OLD_PORT = 54321;
 const EXPECTED_ACCESS_KEY_PROPERTIES =
-    ['id', 'name', 'password', 'port', 'method', 'accessUrl'].sort();
+    ['id', 'name', 'password', 'port', 'method', 'accessUrl', 'dataLimit'].sort();
 
 describe('ShadowsocksManagerService', () => {
   // After processing the response callback, we should set
@@ -66,9 +66,9 @@ describe('ShadowsocksManagerService', () => {
     });
     it('Returns persisted properties', (done) => {
       const repo = getAccessKeyRepository();
-      const accessKeyDataLimit = {bytes: 999};
+      const defaultDataLimit = {bytes: 999};
       const serverConfig =
-          new InMemoryConfig({name: 'Server', accessKeyDataLimit} as ServerConfigJson);
+          new InMemoryConfig({name: 'Server', accessKeyDataLimit: defaultDataLimit} as ServerConfigJson);
       const service = new ShadowsocksManagerServiceBuilder()
                           .serverConfig(serverConfig)
                           .accessKeys(repo)
@@ -78,7 +78,7 @@ describe('ShadowsocksManagerService', () => {
             send: (httpCode, data: ServerInfo) => {
               expect(httpCode).toEqual(200);
               expect(data.name).toEqual('Server');
-              expect(data.accessKeyDataLimit).toEqual(accessKeyDataLimit);
+              expect(data.accessKeyDataLimit).toEqual(defaultDataLimit);
               responseProcessed = true;
             }
           },
@@ -510,10 +510,101 @@ describe('ShadowsocksManagerService', () => {
   });
 
   describe('setAccessKeyDataLimit', () => {
-    it('sets access key limit', async (done) => {
+    it('sets access key data limit', async (done) => {
+      const repo = getAccessKeyRepository();
+      const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
+      const key = await repo.createNewAccessKey();
+      const limit = {bytes: 1000};
+      const res = {send: (httpCode) => {
+        expect(httpCode).toEqual(204);
+        expect(key.dataLimit.bytes).toEqual(1000);
+        responseProcessed = true;
+        done();
+      }};
+      service.setAccessKeyDataLimit({params: {id: key.id, limit}}, res, () => {});
+    });
+
+    it('rejects negative numbers', async (done) => {
+      const repo = getAccessKeyRepository();
+      const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
+      const keyId = (await repo.createNewAccessKey()).id;
+      const limit = {bytes: -1};
+      service.setAccessKeyDataLimit({params: {id: keyId, limit}}, {send: () => {}}, (error) => {
+        expect(error.statusCode).toEqual(400);
+        responseProcessed = true;
+        done();
+      });
+    });
+
+    it('rejects non-numeric limits', async (done) => {
+      const repo = getAccessKeyRepository();
+      const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
+      const keyId = (await repo.createNewAccessKey()).id;
+      const limit = {bytes: "1"};
+      service.setAccessKeyDataLimit({params: {id: keyId, limit}}, {send: () => {}}, (error) => {
+        expect(error.statusCode).toEqual(400);
+        responseProcessed = true;
+        done();
+      });
+    });
+
+    it('rejects an empty request', async (done) => {
+      const repo = getAccessKeyRepository();
+      const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
+      const keyId = (await repo.createNewAccessKey()).id;
+      const limit = {} as DataLimit;
+      service.setAccessKeyDataLimit({params: {id: keyId, limit}}, {send: () => {}}, (error) => {
+        expect(error.statusCode).toEqual(400);
+        responseProcessed = true;
+        done();
+      });
+    });
+
+    it('rejects requests for nonexistent keys', async (done) => {
+      const repo = getAccessKeyRepository();
+      const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
+      await repo.createNewAccessKey();
+      const limit: DataLimit = {bytes: 1000};
+      service.setAccessKeyDataLimit({params: {id: "not an id", limit}}, {send: () => {}}, (error) => {
+        expect(error.statusCode).toEqual(404);
+        responseProcessed = true;
+        done();
+      });
+    });
+  });
+
+  describe('removeAccessKeyDataLimit', () => {
+    it('removes an access key data limit', async (done) => {
+      const repo = getAccessKeyRepository();
+      const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
+      const key = await repo.createNewAccessKey();
+      repo.setAccessKeyDataLimit(key.id, {bytes: 1000});
+      await repo.enforceAccessKeyDataLimits();
+      const res = {send: (httpCode) => {
+        expect(httpCode).toEqual(204);
+        expect(key.dataLimit).toBeFalsy();
+        responseProcessed = true;
+        done();
+      }};
+      service.removeAccessKeyDataLimit({params: {id: key.id}}, res, () => {});
+    });
+    it('returns 404 for a nonexistent key', async (done) => {
+      const repo = getAccessKeyRepository();
+      const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
+      await repo.createNewAccessKey();
+      service.removeAccessKeyDataLimit({params: {id: "not an id"}}, {send: () => {}}, (error) => {
+        expect(error.statusCode).toEqual(404);
+        responseProcessed = true;
+        done();
+      });
+    });
+  });
+
+  describe('setDefaultDataLimit', () => {
+    it('sets default data limit', async (done) => {
       const serverConfig = new InMemoryConfig({} as ServerConfigJson);
       const repo = getAccessKeyRepository();
-      spyOn(repo, 'setAccessKeyDataLimit');
+      spyOn(repo, 'setDefaultDataLimit');
       const service = new ShadowsocksManagerServiceBuilder()
                           .serverConfig(serverConfig)
                           .accessKeys(repo)
@@ -523,7 +614,7 @@ describe('ShadowsocksManagerService', () => {
         send: (httpCode, data) => {
           expect(httpCode).toEqual(204);
           expect(serverConfig.data().accessKeyDataLimit).toEqual(limit);
-          expect(repo.setAccessKeyDataLimit).toHaveBeenCalledWith(limit);
+          expect(repo.setDefaultDataLimit).toHaveBeenCalledWith(limit);
           service.getServer(
               {params: {}}, {
                 send: (httpCode, data: ServerInfo) => {
@@ -535,7 +626,7 @@ describe('ShadowsocksManagerService', () => {
               done);
         }
       };
-      service.setAccessKeyDataLimit({params: {limit}}, res, done);
+      service.setDefaultDataLimit({params: {limit}}, res, done);
     });
     it('returns 400 when limit is missing values', async (done) => {
       const repo = getAccessKeyRepository();
@@ -543,7 +634,7 @@ describe('ShadowsocksManagerService', () => {
       const accessKey = await repo.createNewAccessKey();
       const limit = {} as DataLimit;
       const res = {send: (httpCode, data) => {}};
-      service.setAccessKeyDataLimit({params: {limit}}, res, (error) => {
+      service.setDefaultDataLimit({params: {limit}}, res, (error) => {
         expect(error.statusCode).toEqual(400);
         responseProcessed = true;  // required for afterEach to pass.
         done();
@@ -555,7 +646,7 @@ describe('ShadowsocksManagerService', () => {
       const accessKey = await repo.createNewAccessKey();
       const limit = {bytes: -1};
       const res = {send: (httpCode, data) => {}};
-      service.setAccessKeyDataLimit({params: {limit}}, res, (error) => {
+      service.setDefaultDataLimit({params: {limit}}, res, (error) => {
         expect(error.statusCode).toEqual(400);
         responseProcessed = true;  // required for afterEach to pass.
         done();
@@ -563,12 +654,12 @@ describe('ShadowsocksManagerService', () => {
     });
     it('returns 500 when the repository throws an exception', async (done) => {
       const repo = getAccessKeyRepository();
-      spyOn(repo, 'setAccessKeyDataLimit').and.throwError('cannot write to disk');
+      spyOn(repo, 'setDefaultDataLimit').and.throwError('cannot write to disk');
       const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
       await repo.createNewAccessKey();
       const limit = {bytes: 10000};
       const res = {send: (httpCode, data) => {}};
-      service.setAccessKeyDataLimit({params: {limit}}, res, (error) => {
+      service.setDefaultDataLimit({params: {limit}}, res, (error) => {
         expect(error.statusCode).toEqual(500);
         responseProcessed = true;  // required for afterEach to pass.
         done();
@@ -576,34 +667,34 @@ describe('ShadowsocksManagerService', () => {
     });
   });
 
-  describe('removeAccessKeyDataLimit', () => {
-    it('clears access key limit', async (done) => {
+  describe('removeDefaultDataLimit', () => {
+    it('clears default data limit', async (done) => {
       const limit = {bytes: 10000};
       const serverConfig = new InMemoryConfig({'accessKeyDataLimit': limit} as ServerConfigJson);
       const repo = getAccessKeyRepository();
-      spyOn(repo, 'removeAccessKeyDataLimit').and.callThrough();
+      spyOn(repo, 'removeDefaultDataLimit').and.callThrough();
       const service = new ShadowsocksManagerServiceBuilder()
                           .serverConfig(serverConfig)
                           .accessKeys(repo)
                           .build();
-      await repo.setAccessKeyDataLimit(limit);
+      await repo.setDefaultDataLimit(limit);
       const res = {
         send: (httpCode, data) => {
           expect(httpCode).toEqual(204);
           expect(serverConfig.data().accessKeyDataLimit).toBeUndefined();
-          expect(repo.removeAccessKeyDataLimit).toHaveBeenCalled();
+          expect(repo.removeDefaultDataLimit).toHaveBeenCalled();
           responseProcessed = true;  // required for afterEach to pass.
         }
       };
-      service.removeAccessKeyDataLimit({params: {}}, res, done);
+      service.removeDefaultDataLimit({params: {}}, res, done);
     });
     it('returns 500 when the repository throws an exception', async (done) => {
       const repo = getAccessKeyRepository();
-      spyOn(repo, 'removeAccessKeyDataLimit').and.throwError('cannot write to disk');
+      spyOn(repo, 'removeDefaultDataLimit').and.throwError('cannot write to disk');
       const service = new ShadowsocksManagerServiceBuilder().accessKeys(repo).build();
       const accessKey = await repo.createNewAccessKey();
       const res = {send: (httpCode, data) => {}};
-      service.removeAccessKeyDataLimit({params: {id: accessKey.id}}, res, (error) => {
+      service.removeDefaultDataLimit({params: {id: accessKey.id}}, res, (error) => {
         expect(error.statusCode).toEqual(500);
         responseProcessed = true;  // required for afterEach to pass.
         done();
@@ -712,7 +803,7 @@ function fakeSharedMetricsReporter(): SharedMetricsPublisher {
   };
 }
 
-function getAccessKeyRepository(): AccessKeyRepository {
+function getAccessKeyRepository(): ServerAccessKeyRepository {
   return new ServerAccessKeyRepository(
       OLD_PORT, 'hostname', new InMemoryConfig<AccessKeyConfigJson>({accessKeys: [], nextId: 0}),
       new FakeShadowsocksServer(), new FakePrometheusClient({}));
diff --git a/src/shadowbox/server/manager_service.ts b/src/shadowbox/server/manager_service.ts
index 9c758e34..1f95685c 100644
--- a/src/shadowbox/server/manager_service.ts
+++ b/src/shadowbox/server/manager_service.ts
@@ -28,7 +28,7 @@ import {ServerConfigJson} from './server_config';
 import {SharedMetricsPublisher} from './shared_metrics';
 
 // Creates a AccessKey response.
-function accessKeyToJson(accessKey: AccessKey) {
+function accessKeyToApiJson(accessKey: AccessKey) {
   return {
     // The unique identifier of this access key.
     id: accessKey.id,
@@ -38,12 +38,13 @@ function accessKeyToJson(accessKey: AccessKey) {
     password: accessKey.proxyParams.password,
     port: accessKey.proxyParams.portNumber,
     method: accessKey.proxyParams.encryptionMethod,
+    dataLimit: accessKey.dataLimit,
     accessUrl: SIP002_URI.stringify(makeConfig({
       host: accessKey.proxyParams.hostname,
       port: accessKey.proxyParams.portNumber,
       method: accessKey.proxyParams.encryptionMethod,
       password: accessKey.proxyParams.password,
-      outline: 1,
+      outline: 1
     }))
   };
 }
@@ -78,9 +79,9 @@ export function bindService(
   apiServer.put(`${apiPrefix}/name`, service.renameServer.bind(service));
   apiServer.get(`${apiPrefix}/server`, service.getServer.bind(service));
   apiServer.put(
-      `${apiPrefix}/server/access-key-data-limit`, service.setAccessKeyDataLimit.bind(service));
+      `${apiPrefix}/server/access-key-data-limit`, service.setDefaultDataLimit.bind(service));
   apiServer.del(
-      `${apiPrefix}/server/access-key-data-limit`, service.removeAccessKeyDataLimit.bind(service));
+      `${apiPrefix}/server/access-key-data-limit`, service.removeDefaultDataLimit.bind(service));
   apiServer.put(
       `${apiPrefix}/server/hostname-for-access-keys`,
       service.setHostnameForAccessKeys.bind(service));
@@ -93,6 +94,8 @@ export function bindService(
 
   apiServer.del(`${apiPrefix}/access-keys/:id`, service.removeAccessKey.bind(service));
   apiServer.put(`${apiPrefix}/access-keys/:id/name`, service.renameAccessKey.bind(service));
+  apiServer.put(`${apiPrefix}/access-keys/:id/data-limit`, service.setAccessKeyDataLimit.bind(service));
+  apiServer.del(`${apiPrefix}/access-keys/:id/data-limit`, service.removeAccessKeyDataLimit.bind(service));
 
   apiServer.get(`${apiPrefix}/metrics/transfer`, service.getDataUsage.bind(service));
   apiServer.get(`${apiPrefix}/metrics/enabled`, service.getShareMetrics.bind(service));
@@ -125,6 +128,19 @@ function validateAccessKeyId(accessKeyId: unknown): string {
   return accessKeyId;
 }
 
+function validateDataLimit(limit: unknown): DataLimit {
+  if (!limit) {
+    throw new restifyErrors.MissingParameterError(
+        {statusCode: 400}, 'Missing `limit` parameter');
+  }
+  const bytes = (limit as DataLimit).bytes;
+  if (!(Number.isInteger(bytes) && bytes >= 0)) {
+    throw new restifyErrors.InvalidArgumentError(
+        {statusCode: 400}, '`limit.bytes` must be an non-negative integer');
+  }
+  return limit as DataLimit;
+}
+
 // The ShadowsocksManagerService manages the access keys that can use the server
 // as a proxy using Shadowsocks. It runs an instance of the Shadowsocks server
 // for each existing access key, with the port and password assigned for that access key.
@@ -201,7 +217,7 @@ export class ShadowsocksManagerService {
     logging.debug(`listAccessKeys request ${JSON.stringify(req.params)}`);
     const response = {accessKeys: []};
     for (const accessKey of this.accessKeys.listAccessKeys()) {
-      response.accessKeys.push(accessKeyToJson(accessKey));
+      response.accessKeys.push(accessKeyToApiJson(accessKey));
     }
     logging.debug(`listAccessKeys response ${JSON.stringify(response)}`);
     res.send(HttpSuccess.OK, response);
@@ -213,7 +229,7 @@ export class ShadowsocksManagerService {
     try {
       logging.debug(`createNewAccessKey request ${JSON.stringify(req.params)}`);
       this.accessKeys.createNewAccessKey().then((accessKey) => {
-        const accessKeyJson = accessKeyToJson(accessKey);
+        const accessKeyJson = accessKeyToApiJson(accessKey);
         res.send(201, accessKeyJson);
         logging.debug(`createNewAccessKey response ${JSON.stringify(accessKeyJson)}`);
         return next();
@@ -302,32 +318,66 @@ export class ShadowsocksManagerService {
   public async setAccessKeyDataLimit(req: RequestType, res: ResponseType, next: restify.Next) {
     try {
       logging.debug(`setAccessKeyDataLimit request ${JSON.stringify(req.params)}`);
-      const limit = req.params.limit as DataLimit;
-      if (!limit) {
-        return next(new restifyErrors.MissingParameterError(
-            {statusCode: 400}, 'Missing `limit` parameter'));
-      } else if (!Number.isInteger(limit.bytes)) {
-        return next(new restifyErrors.InvalidArgumentError(
-            {statusCode: 400}, '`limit` must be an integer'));
-      }
-      this.accessKeys.setAccessKeyDataLimit(limit);
-      this.serverConfig.data().accessKeyDataLimit = limit;
-      this.serverConfig.write();
+      const accessKeyId = validateAccessKeyId(req.params.id);
+      const limit = validateDataLimit(req.params.limit);
+      // Enforcement is done asynchronously in the proxy server.  This is transparent to the manager
+      // so this doesn't introduce any race conditions between the server and UI.
+      this.accessKeys.setAccessKeyDataLimit(accessKeyId, limit);
       res.send(HttpSuccess.NO_CONTENT);
       return next();
-    } catch (error) {
+    } catch(error) {
       logging.error(error);
-      if (error instanceof errors.InvalidAccessKeyDataLimit) {
-        return next(new restifyErrors.InvalidArgumentError({statusCode: 400}, error.message));
+      if (error instanceof errors.AccessKeyNotFound) {
+        return next(new restifyErrors.NotFoundError(error.message));
       }
-      return next(new restifyErrors.InternalServerError());
+      return next(error);
     }
   }
 
   public async removeAccessKeyDataLimit(req: RequestType, res: ResponseType, next: restify.Next) {
     try {
       logging.debug(`removeAccessKeyDataLimit request ${JSON.stringify(req.params)}`);
-      await this.accessKeys.removeAccessKeyDataLimit();
+      const accessKeyId = validateAccessKeyId(req.params.id);
+      // Enforcement is done asynchronously in the proxy server.  This is transparent to the manager
+      // so this doesn't introduce any race conditions between the server and UI.
+      this.accessKeys.removeAccessKeyDataLimit(accessKeyId);
+      res.send(HttpSuccess.NO_CONTENT);
+      return next();
+    } catch(error) {
+      logging.error(error);
+      if (error instanceof errors.AccessKeyNotFound) {
+        return next(new restifyErrors.NotFoundError(error.message));
+      }
+      return next(error);
+    }
+  }
+
+  public async setDefaultDataLimit(req: RequestType, res: ResponseType, next: restify.Next) {
+    try {
+      logging.debug(`setDefaultDataLimit request ${JSON.stringify(req.params)}`);
+      const limit = validateDataLimit(req.params.limit);
+      // Enforcement is done asynchronously in the proxy server.  This is transparent to the manager
+      // so this doesn't introduce any race conditions between the server and UI.
+      this.accessKeys.setDefaultDataLimit(limit);
+      this.serverConfig.data().accessKeyDataLimit = limit;
+      this.serverConfig.write();
+      res.send(HttpSuccess.NO_CONTENT);
+      return next();
+    } catch (error) {
+      logging.error(error);
+      if (error instanceof restifyErrors.InvalidArgumentError || error instanceof restifyErrors.MissingParameterError) {
+        return next(error);
+      }
+      return next(new restifyErrors.InternalServerError());
+    }
+  }
+
+  public async removeDefaultDataLimit(req: RequestType, res: ResponseType, next: restify.Next) {
+    try {
+      logging.debug(`removeDefaultDataLimit request ${JSON.stringify(req.params)}`);
+      // Enforcement is done asynchronously in the proxy server.  This is transparent to the manager
+      // so this doesn't introduce any race conditions between the server and UI.
+      this.accessKeys.removeDefaultDataLimit();
       delete this.serverConfig.data().accessKeyDataLimit;
       this.serverConfig.write();
       res.send(HttpSuccess.NO_CONTENT);
diff --git a/src/shadowbox/server/server_access_key.spec.ts b/src/shadowbox/server/server_access_key.spec.ts
index 8225188f..531cd1c8 100644
--- a/src/shadowbox/server/server_access_key.spec.ts
+++ b/src/shadowbox/server/server_access_key.spec.ts
@@ -17,7 +17,7 @@ import * as net from 'net';
 import {ManualClock} from '../infrastructure/clock';
 import {PortProvider} from '../infrastructure/get_port';
 import {InMemoryConfig} from '../infrastructure/json_config';
-import {AccessKeyRepository, DataLimit} from '../model/access_key';
+import {AccessKey, AccessKeyId, AccessKeyRepository, DataLimit} from '../model/access_key';
 import * as errors from '../model/errors';
 
 import {FakePrometheusClient, FakeShadowsocksServer} from './mocks/mocks';
@@ -164,34 +164,131 @@ describe('ServerAccessKeyRepository', () => {
       done();
     });
   });
+  
+  it('setAccessKeyDataLimit can set a custom data limit', async(done) => {
+    const server = new FakeShadowsocksServer();
+    const config = new InMemoryConfig<AccessKeyConfigJson>({accessKeys: [], nextId: 0});
+    const repo = new RepoBuilder().shadowsocksServer(server).keyConfig(config).build();
+    const key = await repo.createNewAccessKey();
+    const limit = {bytes: 5000};
+    await expectNoAsyncThrow(repo.setAccessKeyDataLimit.bind(repo, key.id, {bytes: 5000}));
+    expect(key.dataLimit).toEqual(limit);
+    expect(config.mostRecentWrite.accessKeys[0].dataLimit).toEqual(limit);
+    done();
+  });
 
-  it('can set access key data limit', async (done) => {
+  async function setKeyLimitAndEnforce(
+      repo: ServerAccessKeyRepository, id: AccessKeyId, limit: DataLimit) {
+    repo.setAccessKeyDataLimit(id, limit);
+    // We enforce asynchronously, in setAccessKeyDataLimit, so explicitly call it here to make sure
+    // enforcement is done before we make assertions.
+    return repo.enforceAccessKeyDataLimits();
+  }
+
+  it('setAccessKeyDataLimit can change a key\'s limit status', async(done) => {
+    const server = new FakeShadowsocksServer();
+    const prometheusClient = new FakePrometheusClient({'0': 500});
+    const repo = 
+      new RepoBuilder().prometheusClient(prometheusClient).shadowsocksServer(server).build();
+    await repo.start(new ManualClock());
+    const key = await repo.createNewAccessKey();
+    await setKeyLimitAndEnforce(repo, key.id, {bytes: 0});
+
+    expect(key.isOverDataLimit).toBeTruthy();
+    let serverKeys = server.getAccessKeys();
+    expect(serverKeys.length).toEqual(0);
+
+    await setKeyLimitAndEnforce(repo, key.id, {bytes: 1000});
+
+    expect(key.isOverDataLimit).toBeFalsy();
+    serverKeys = server.getAccessKeys();
+    expect(serverKeys.length).toEqual(1);
+    expect(serverKeys[0].id).toEqual(key.id);
+    done();
+  });
+  
+  it('setAccessKeyDataLimit overrides default data limit', async(done) => {
+    const server = new FakeShadowsocksServer();
+    const prometheusClient = new FakePrometheusClient({'0': 750, '1': 1250});
+    const repo = 
+      new RepoBuilder().prometheusClient(prometheusClient).shadowsocksServer(server).build();  
+    await repo.start(new ManualClock());
+    const lowerLimitThanDefault = await repo.createNewAccessKey();
+    const higherLimitThanDefault = await repo.createNewAccessKey();
+    await repo.setDefaultDataLimit({bytes: 1000});
+
+    expect(lowerLimitThanDefault.isOverDataLimit).toBeFalsy();
+    await setKeyLimitAndEnforce(repo, lowerLimitThanDefault.id, {bytes: 500});
+    expect(lowerLimitThanDefault.isOverDataLimit).toBeTruthy();
+
+    expect(higherLimitThanDefault.isOverDataLimit).toBeTruthy();
+    await setKeyLimitAndEnforce(repo, higherLimitThanDefault.id, {bytes: 1500});
+    expect(higherLimitThanDefault.isOverDataLimit).toBeFalsy();
+    done();
+  });
+
+  async function removeKeyLimitAndEnforce(repo: ServerAccessKeyRepository, id: AccessKeyId) {
+    repo.removeAccessKeyDataLimit(id);
+    // We enforce asynchronously, in setAccessKeyDataLimit, so explicitly call it here to make sure
+    // enforcement is done before we make assertions.
+    return repo.enforceAccessKeyDataLimits();
+  }
+
+  it('removeAccessKeyDataLimit can remove a custom data limit', async(done) => {
+    const server = new FakeShadowsocksServer();
+    const config = new InMemoryConfig<AccessKeyConfigJson>({accessKeys: [], nextId: 0});
+    const repo = new RepoBuilder().shadowsocksServer(server).keyConfig(config).build();
+    const key = await repo.createNewAccessKey();
+    await setKeyLimitAndEnforce(repo, key.id, {bytes: 1});
+    await expectNoAsyncThrow(repo.removeAccessKeyDataLimit.bind(repo, key.id));
+    expect(key.dataLimit).toBeFalsy();
+    expect(config.mostRecentWrite.accessKeys[0].dataLimit).not.toBeDefined();
+    done();
+  });
+  
+  it('removeAccessKeyDataLimit restores a key to the default data limit', async(done) => {
+    const server = new FakeShadowsocksServer();
+    const prometheusClient = new FakePrometheusClient({'0': 500});
+    const repo = 
+      new RepoBuilder().prometheusClient(prometheusClient).shadowsocksServer(server).build();
+    const key = await repo.createNewAccessKey();
+    await repo.start(new ManualClock());
+    await repo.setDefaultDataLimit({bytes: 0});
+    await setKeyLimitAndEnforce(repo, key.id, {bytes: 1000});
+    expect(key.isOverDataLimit).toBeFalsy();
+
+    await removeKeyLimitAndEnforce(repo, key.id);
+    expect(key.isOverDataLimit).toBeTruthy();
+    done();
+  });
+  
+  it('removeAccessKeyDataLimit can restore an over-limit access key', async(done) => {
+    const server = new FakeShadowsocksServer();
+    const prometheusClient = new FakePrometheusClient({'0': 500});
+    const repo = 
+      new RepoBuilder().prometheusClient(prometheusClient).shadowsocksServer(server).build();
+    const key = await repo.createNewAccessKey();
+    await repo.start(new ManualClock());
+
+    await setKeyLimitAndEnforce(repo, key.id, {bytes: 0});
+    expect(key.isOverDataLimit).toBeTruthy();
+    expect(server.getAccessKeys().length).toEqual(0);
+
+    await removeKeyLimitAndEnforce(repo, key.id);
+    expect(key.isOverDataLimit).toBeFalsy();
+    expect(server.getAccessKeys().length).toEqual(1);
+    done();
+  });
+  
+  it('can set default data limit', async (done) => {
     const repo = new RepoBuilder().build();
     const limit = {bytes: 5000};
-    await expectNoAsyncThrow(repo.setAccessKeyDataLimit.bind(repo, limit));
-    expect(repo.dataLimit).toEqual(limit);
+    await expectNoAsyncThrow(repo.setDefaultDataLimit.bind(repo, limit));
+    expect(repo.defaultDataLimit).toEqual(limit);
     done();
   });
 
-  it('setAccessKeyDataLimit fails with disallowed limit values', async (done) => {
-    const repo = new RepoBuilder().build();
-    const accessKey = await repo.createNewAccessKey();
-    // Negative values
-    const negativeBytesLimit = {bytes: -1000};
-    await expectAsyncThrow(
-        repo.setAccessKeyDataLimit.bind(repo, negativeBytesLimit),
-        errors.InvalidAccessKeyDataLimit);
-    // Missing properties
-    const missingDataLimit = {} as DataLimit;
-    await expectAsyncThrow(
-        repo.setAccessKeyDataLimit.bind(repo, missingDataLimit), errors.InvalidAccessKeyDataLimit);
-    // Undefined limit
-    await expectAsyncThrow(
-        repo.setAccessKeyDataLimit.bind(repo, undefined), errors.InvalidAccessKeyDataLimit);
-    done();
-  });
-
-  it('setAccessKeyDataLimit updates keys limit status', async (done) => {
+  it('setDefaultDataLimit updates keys limit status', async (done) => {
     const server = new FakeShadowsocksServer();
     const prometheusClient = new FakePrometheusClient({'0': 500, '1': 200});
     const repo =
@@ -200,7 +297,10 @@ describe('ServerAccessKeyRepository', () => {
     const accessKey2 = await repo.createNewAccessKey();
     await repo.start(new ManualClock());
 
-    await repo.setAccessKeyDataLimit({bytes: 250});
+    repo.setDefaultDataLimit({bytes: 250});
+    // We enforce asynchronously, in setAccessKeyDataLimit, so explicitly call it here to make sure
+    // enforcement is done before we make assertions.
+    await repo.enforceAccessKeyDataLimits();
     expect(accessKey1.isOverDataLimit).toBeTruthy();
     expect(accessKey2.isOverDataLimit).toBeFalsy();
     // We determine which access keys have been enabled/disabled by accessing them from
@@ -211,7 +311,8 @@ describe('ServerAccessKeyRepository', () => {
     // The over-limit key should be re-enabled after increasing the data limit, while the other key
     // should be disabled after its data usage increased.
     prometheusClient.bytesTransferredById = {'0': 500, '1': 1000};
-    await repo.setAccessKeyDataLimit({bytes: 700});
+    repo.setDefaultDataLimit({bytes: 700});
+    await repo.enforceAccessKeyDataLimits();
     expect(accessKey1.isOverDataLimit).toBeFalsy();
     expect(accessKey2.isOverDataLimit).toBeTruthy();
     serverAccessKeys = server.getAccessKeys();
@@ -220,22 +321,22 @@ describe('ServerAccessKeyRepository', () => {
     done();
   });
 
-  it('can remove access key limits', async (done) => {
+  it('can remove the default data limit', async (done) => {
     const limit = {bytes: 100};
-    const repo = new RepoBuilder().accessKeyDataLimit(limit).build();
-    expect(repo.dataLimit).toEqual(limit);
-    await expectNoAsyncThrow(repo.removeAccessKeyDataLimit.bind(repo));
-    expect(repo.dataLimit).toBeUndefined();
+    const repo = new RepoBuilder().defaultDataLimit(limit).build();
+    expect(repo.defaultDataLimit).toEqual(limit);
+    await expectNoAsyncThrow(repo.removeDefaultDataLimit.bind(repo));
+    expect(repo.defaultDataLimit).toBeUndefined();
     done();
   });
 
-  it('removeAccessKeyDataLimit restores over-limit access keys', async (done) => {
+  it('removeDefaultDataLimit restores over-limit access keys', async (done) => {
     const server = new FakeShadowsocksServer();
     const prometheusClient = new FakePrometheusClient({'0': 500, '1': 100});
     const repo = new RepoBuilder()
                      .prometheusClient(prometheusClient)
                      .shadowsocksServer(server)
-                     .accessKeyDataLimit({bytes: 200})
+                     .defaultDataLimit({bytes: 200})
                      .build();
 
     const accessKey1 = await repo.createNewAccessKey();
@@ -244,7 +345,10 @@ describe('ServerAccessKeyRepository', () => {
     expect(server.getAccessKeys().length).toEqual(1);
 
     // Remove the limit; expect the key to be under limit and enabled.
-    await expectNoAsyncThrow(repo.removeAccessKeyDataLimit.bind(repo));
+    expectNoAsyncThrow(repo.removeDefaultDataLimit.bind(repo));
+    // We enforce asynchronously, in setAccessKeyDataLimit, so explicitly call it here to make sure
+    // enforcement is done before we make assertions.
+    await repo.enforceAccessKeyDataLimits();
     expect(server.getAccessKeys().length).toEqual(2);
     expect(accessKey1.isOverDataLimit).toBeFalsy();
     expect(accessKey2.isOverDataLimit).toBeFalsy();
@@ -256,7 +360,7 @@ describe('ServerAccessKeyRepository', () => {
         new FakePrometheusClient({'0': 100, '1': 200, '2': 300, '3': 400, '4': 500});
     const limit = {bytes: 250};
     const repo =
-        new RepoBuilder().prometheusClient(prometheusClient).accessKeyDataLimit(limit).build();
+        new RepoBuilder().prometheusClient(prometheusClient).defaultDataLimit(limit).build();
     for (let i = 0; i < Object.keys(prometheusClient.bytesTransferredById).length; ++i) {
       await repo.createNewAccessKey();
     }
@@ -276,13 +380,34 @@ describe('ServerAccessKeyRepository', () => {
     done();
   });
 
+  it('enforceAccessKeyDataLimits respects both default and per-key limits', async (done) => {
+    const prometheusClient = new FakePrometheusClient({'0': 200, '1': 300});
+    const repo =
+        new RepoBuilder().prometheusClient(prometheusClient).defaultDataLimit({bytes: 500}).build();
+    const perKeyLimited = await repo.createNewAccessKey();
+    const defaultLimited = await repo.createNewAccessKey();
+    await setKeyLimitAndEnforce(repo, perKeyLimited.id, {bytes: 100});
+
+    await repo.enforceAccessKeyDataLimits();
+    expect(perKeyLimited.isOverDataLimit).toBeTruthy();
+    expect(defaultLimited.isOverDataLimit).toBeFalsy();
+
+    prometheusClient.bytesTransferredById[perKeyLimited.id] = 50;
+    prometheusClient.bytesTransferredById[defaultLimited.id] = 600;
+    await repo.enforceAccessKeyDataLimits();
+    expect(perKeyLimited.isOverDataLimit).toBeFalsy();
+    expect(defaultLimited.isOverDataLimit).toBeTruthy();
+
+    done();
+  });
+
   it('enforceAccessKeyDataLimits enables and disables keys', async (done) => {
     const server = new FakeShadowsocksServer();
     const prometheusClient = new FakePrometheusClient({'0': 500, '1': 100});
     const repo = new RepoBuilder()
                      .prometheusClient(prometheusClient)
                      .shadowsocksServer(server)
-                     .accessKeyDataLimit({bytes: 200})
+                     .defaultDataLimit({bytes: 200})
                      .build();
 
     const accessKey1 = await repo.createNewAccessKey();
@@ -361,7 +486,7 @@ describe('ServerAccessKeyRepository', () => {
     const accessKey1 = await repo.createNewAccessKey();
     const accessKey2 = await repo.createNewAccessKey();
     const accessKey3 = await repo.createNewAccessKey();
-    await repo.setAccessKeyDataLimit({bytes: 300});
+    await repo.setDefaultDataLimit({bytes: 300});
     const clock = new ManualClock();
     await repo.start(clock);
     await clock.runCallbacks();
@@ -429,7 +554,7 @@ class RepoBuilder {
   private keyConfig_ = new InMemoryConfig<AccessKeyConfigJson>({accessKeys: [], nextId: 0});
   private shadowsocksServer_ = new FakeShadowsocksServer();
   private prometheusClient_ = new FakePrometheusClient({});
-  private accessKeyDataLimit_;
+  private defaultDataLimit_;
 
   public port(port: number): RepoBuilder {
     this.port_ = port;
@@ -447,14 +572,14 @@ class RepoBuilder {
     this.prometheusClient_ = prometheusClient;
     return this;
   }
-  public accessKeyDataLimit(limit: DataLimit): RepoBuilder {
-    this.accessKeyDataLimit_ = limit;
+  public defaultDataLimit(limit: DataLimit): RepoBuilder {
+    this.defaultDataLimit_ = limit;
     return this;
   }
 
   public build(): ServerAccessKeyRepository {
     return new ServerAccessKeyRepository(
         this.port_, 'hostname', this.keyConfig_, this.shadowsocksServer_, this.prometheusClient_,
-        this.accessKeyDataLimit_);
+        this.defaultDataLimit_);
   }
 }
diff --git a/src/shadowbox/server/server_access_key.ts b/src/shadowbox/server/server_access_key.ts
index b2166714..53e5c87c 100644
--- a/src/shadowbox/server/server_access_key.ts
+++ b/src/shadowbox/server/server_access_key.ts
@@ -26,18 +26,19 @@ import {ShadowsocksServer} from '../model/shadowsocks_server';
 import {PrometheusManagerMetrics} from './manager_metrics';
 
 // The format as json of access keys in the config file.
-interface AccessKeyJson {
+interface AccessKeyStorageJson {
   id: AccessKeyId;
   metricsId: AccessKeyId;
   name: string;
   password: string;
   port: number;
   encryptionMethod?: string;
+  dataLimit?: DataLimit;
 }
 
 // The configuration file format as json.
 export interface AccessKeyConfigJson {
-  accessKeys?: AccessKeyJson[];
+  accessKeys?: AccessKeyStorageJson[];
   // Next AccessKeyId to use.
   nextId?: number;
 }
@@ -47,11 +48,7 @@ class ServerAccessKey implements AccessKey {
   public isOverDataLimit = false;
   constructor(
       readonly id: AccessKeyId, public name: string, public metricsId: AccessKeyMetricsId,
-      readonly proxyParams: ProxyParams) {}
-}
-
-function isValidAccessKeyDataLimit(limit: DataLimit): boolean {
-  return limit && limit.bytes >= 0;
+      readonly proxyParams: ProxyParams, public dataLimit?: DataLimit) {}
 }
 
 // Generates a random password for Shadowsocks access keys.
@@ -59,7 +56,7 @@ function generatePassword(): string {
   return randomstring.generate(12);
 }
 
-function makeAccessKey(hostname: string, accessKeyJson: AccessKeyJson): AccessKey {
+function makeAccessKey(hostname: string, accessKeyJson: AccessKeyStorageJson): AccessKey {
   const proxyParams = {
     hostname,
     portNumber: accessKeyJson.port,
@@ -70,14 +67,15 @@ function makeAccessKey(hostname: string, accessKeyJson: AccessKeyJson): AccessKe
       accessKeyJson.id, accessKeyJson.name, accessKeyJson.metricsId, proxyParams);
 }
 
-function makeAccessKeyJson(accessKey: AccessKey): AccessKeyJson {
+function accessKeyToStorageJson(accessKey: AccessKey): AccessKeyStorageJson {
   return {
     id: accessKey.id,
     metricsId: accessKey.metricsId,
     name: accessKey.name,
     password: accessKey.proxyParams.password,
     port: accessKey.proxyParams.portNumber,
-    encryptionMethod: accessKey.proxyParams.encryptionMethod
+    encryptionMethod: accessKey.proxyParams.encryptionMethod,
+    dataLimit: accessKey.dataLimit
   };
 }
 
@@ -93,7 +91,7 @@ export class ServerAccessKeyRepository implements AccessKeyRepository {
       private portForNewAccessKeys: number, private proxyHostname: string,
       private keyConfig: JsonConfig<AccessKeyConfigJson>,
       private shadowsocksServer: ShadowsocksServer, private prometheusClient: PrometheusClient,
-      private accessKeyDataLimit?: DataLimit) {
+      private _defaultDataLimit?: DataLimit) {
     if (this.keyConfig.data().accessKeys === undefined) {
       this.keyConfig.data().accessKeys = [];
     }
@@ -180,21 +178,31 @@ export class ServerAccessKeyRepository implements AccessKeyRepository {
     this.saveAccessKeys();
   }
 
-  get dataLimit(): DataLimit|undefined {
-    return this.accessKeyDataLimit;
+
+  setAccessKeyDataLimit(id: AccessKeyId, limit: DataLimit): void {
+    this.getAccessKey(id).dataLimit = limit;
+    this.saveAccessKeys();
+    this.enforceAccessKeyDataLimits();
   }
 
-  setAccessKeyDataLimit(limit: DataLimit): Promise<void> {
-    if (!isValidAccessKeyDataLimit(limit)) {
-      throw new errors.InvalidAccessKeyDataLimit();
-    }
-    this.accessKeyDataLimit = limit;
-    return this.enforceAccessKeyDataLimits();
+  removeAccessKeyDataLimit(id: AccessKeyId): void {
+    delete this.getAccessKey(id).dataLimit;
+    this.saveAccessKeys();
+    this.enforceAccessKeyDataLimits();
   }
 
-  removeAccessKeyDataLimit(): Promise<void> {
-    delete this.accessKeyDataLimit;
-    return this.enforceAccessKeyDataLimits();
+  get defaultDataLimit(): DataLimit|undefined {
+    return this._defaultDataLimit;
+  }
+
+  setDefaultDataLimit(limit: DataLimit): void {
+    this._defaultDataLimit = limit;
+    this.enforceAccessKeyDataLimits();
+  }
+
+  removeDefaultDataLimit(): void {
+    delete this._defaultDataLimit;
+    this.enforceAccessKeyDataLimits();
   }
 
   getMetricsId(id: AccessKeyId): AccessKeyMetricsId|undefined {
@@ -210,10 +218,13 @@ export class ServerAccessKeyRepository implements AccessKeyRepository {
         (await metrics.getOutboundByteTransfer({hours: 30 * 24})).bytesTransferredByUserId;
     let limitStatusChanged = false;
     for (const accessKey of this.accessKeys) {
-      const usageBytes = bytesTransferredById[accessKey.id] || 0;
+      const usageBytes = bytesTransferredById[accessKey.id] ?? 0;
       const wasOverDataLimit = accessKey.isOverDataLimit;
-      accessKey.isOverDataLimit =
-          this.accessKeyDataLimit ? usageBytes > this.accessKeyDataLimit.bytes : false;
+      let limitBytes = (accessKey.dataLimit ?? this._defaultDataLimit)?.bytes;
+      if (limitBytes === undefined) {
+        limitBytes = Number.POSITIVE_INFINITY;
+      }
+      accessKey.isOverDataLimit = usageBytes > limitBytes;
       limitStatusChanged = accessKey.isOverDataLimit !== wasOverDataLimit || limitStatusChanged;
     }
     if (limitStatusChanged) {
@@ -238,7 +249,7 @@ export class ServerAccessKeyRepository implements AccessKeyRepository {
   }
 
   private saveAccessKeys() {
-    this.keyConfig.data().accessKeys = this.accessKeys.map(key => makeAccessKeyJson(key));
+    this.keyConfig.data().accessKeys = this.accessKeys.map(key => accessKeyToStorageJson(key));
     this.keyConfig.write();
   }
 
diff --git a/src/shadowbox/server/server_config.ts b/src/shadowbox/server/server_config.ts
index 6cbd58b1..2e757abe 100644
--- a/src/shadowbox/server/server_config.ts
+++ b/src/shadowbox/server/server_config.ts
@@ -35,7 +35,7 @@ export interface ServerConfigJson {
   // We don't serialize the shadowbox version, this is obtained dynamically from node.
   // Public proxy hostname.
   hostname?: string;
-  // Data transfer limit applied to all access keys.
+  // Default data transfer limit applied to all access keys.
   accessKeyDataLimit?: DataLimit;
 }
 
diff --git a/src/shadowbox/server/shared_metrics.spec.ts b/src/shadowbox/server/shared_metrics.spec.ts
index fe62ef7b..6b2fb03b 100644
--- a/src/shadowbox/server/shared_metrics.spec.ts
+++ b/src/shadowbox/server/shared_metrics.spec.ts
@@ -14,8 +14,9 @@
 
 import {ManualClock} from '../infrastructure/clock';
 import {InMemoryConfig} from '../infrastructure/json_config';
-import {AccessKeyId} from '../model/access_key';
+import {AccessKeyId, DataLimit} from '../model/access_key';
 import {version} from '../package.json';
+import {AccessKeyConfigJson} from './server_access_key';
 
 import {ServerConfigJson} from './server_config';
 import {DailyFeatureMetricsReportJson, HourlyServerMetricsReportJson, KeyUsage, MetricsCollectorClient, OutlineSharedMetricsPublisher, UsageMetrics} from './shared_metrics';
@@ -26,7 +27,7 @@ describe('OutlineSharedMetricsPublisher', () => {
       const serverConfig = new InMemoryConfig<ServerConfigJson>({});
 
       const publisher =
-          new OutlineSharedMetricsPublisher(new ManualClock(), serverConfig, null, null, null);
+          new OutlineSharedMetricsPublisher(new ManualClock(), serverConfig, null, null, null, null);
       expect(publisher.isSharingEnabled()).toBeFalsy();
 
       publisher.startSharing();
@@ -40,7 +41,7 @@ describe('OutlineSharedMetricsPublisher', () => {
     it('Reads from config', () => {
       const serverConfig = new InMemoryConfig<ServerConfigJson>({metricsEnabled: true});
       const publisher =
-          new OutlineSharedMetricsPublisher(new ManualClock(), serverConfig, null, null, null);
+          new OutlineSharedMetricsPublisher(new ManualClock(), serverConfig, null, null, null, null);
       expect(publisher.isSharingEnabled()).toBeTruthy();
     });
   });
@@ -53,7 +54,7 @@ describe('OutlineSharedMetricsPublisher', () => {
       const toMetricsId = (id: AccessKeyId) => `M(${id})`;
       const metricsCollector = new FakeMetricsCollector();
       const publisher = new OutlineSharedMetricsPublisher(
-          clock, serverConfig, usageMetrics, toMetricsId, metricsCollector);
+          clock, serverConfig, null, usageMetrics, toMetricsId, metricsCollector);
 
       publisher.startSharing();
       usageMetrics.usage = [
@@ -103,7 +104,7 @@ describe('OutlineSharedMetricsPublisher', () => {
       const toMetricsId = (id: AccessKeyId) => `M(${id})`;
       const metricsCollector = new FakeMetricsCollector();
       const publisher = new OutlineSharedMetricsPublisher(
-          clock, serverConfig, usageMetrics, toMetricsId, metricsCollector);
+          clock, serverConfig, null, usageMetrics, toMetricsId, metricsCollector);
 
       publisher.startSharing();
       usageMetrics.usage = [
@@ -131,9 +132,26 @@ describe('OutlineSharedMetricsPublisher', () => {
     let timestamp = clock.nowMs;
     const serverConfig = new InMemoryConfig<ServerConfigJson>(
         {serverId: 'server-id', accessKeyDataLimit: {bytes: 123}});
+    let keyId = 0;
+    const makeKeyJson = (dataLimit?: DataLimit) => {
+      return {
+        id: (keyId++).toString(),
+        metricsId: "id",
+        name: "name",
+        password: "pass",
+        port: 12345,
+        dataLimit,
+      };
+    };
+    const keyConfig = new InMemoryConfig<AccessKeyConfigJson>({
+      accessKeys: [
+        makeKeyJson({bytes: 2}),
+        makeKeyJson()
+      ]
+    });
     const metricsCollector = new FakeMetricsCollector();
     const publisher = new OutlineSharedMetricsPublisher(
-        clock, serverConfig, new ManualUsageMetrics(), (id: AccessKeyId) => '', metricsCollector);
+        clock, serverConfig, keyConfig, new ManualUsageMetrics(), (id: AccessKeyId) => '', metricsCollector);
 
     publisher.startSharing();
     await clock.runCallbacks();
@@ -141,7 +159,10 @@ describe('OutlineSharedMetricsPublisher', () => {
       serverId: 'server-id',
       serverVersion: version,
       timestampUtcMs: timestamp,
-      dataLimit: {enabled: true}
+      dataLimit: {
+        enabled: true,
+        perKeyLimitCount: 1
+      }
     });
     clock.nowMs += 24 * 60 * 60 * 1000;
     timestamp = clock.nowMs;
@@ -152,8 +173,16 @@ describe('OutlineSharedMetricsPublisher', () => {
       serverId: 'server-id',
       serverVersion: version,
       timestampUtcMs: timestamp,
-      dataLimit: {enabled: false}
+      dataLimit: {
+        enabled: false,
+        perKeyLimitCount: 1
+      }
     });
+
+    clock.nowMs += 24 * 60 * 60 * 1000;
+    delete keyConfig.data().accessKeys[0].dataLimit;
+    await clock.runCallbacks();
+    expect(metricsCollector.collectedFeatureMetricsReport.dataLimit.perKeyLimitCount).toEqual(0);
   });
   it('does not report metrics when sharing is disabled', async () => {
     const clock = new ManualClock();
@@ -163,7 +192,7 @@ describe('OutlineSharedMetricsPublisher', () => {
     spyOn(metricsCollector, 'collectServerUsageMetrics').and.callThrough();
     spyOn(metricsCollector, 'collectFeatureMetrics').and.callThrough();
     const publisher = new OutlineSharedMetricsPublisher(
-        clock, serverConfig, new ManualUsageMetrics(), (id: AccessKeyId) => '', metricsCollector);
+        clock, serverConfig, new InMemoryConfig<AccessKeyConfigJson>({}), new ManualUsageMetrics(), (id: AccessKeyId) => '', metricsCollector);
 
     await clock.runCallbacks();
     expect(metricsCollector.collectServerUsageMetrics).not.toHaveBeenCalled();
@@ -192,4 +221,4 @@ class ManualUsageMetrics implements UsageMetrics {
   reset() {
     this.usage = [] as KeyUsage[];
   }
-}
\ No newline at end of file
+}
diff --git a/src/shadowbox/server/shared_metrics.ts b/src/shadowbox/server/shared_metrics.ts
index 53909365..9f865970 100644
--- a/src/shadowbox/server/shared_metrics.ts
+++ b/src/shadowbox/server/shared_metrics.ts
@@ -20,6 +20,7 @@ import * as logging from '../infrastructure/logging';
 import {PrometheusClient} from '../infrastructure/prometheus_scraper';
 import {AccessKeyId, AccessKeyMetricsId} from '../model/access_key';
 import {version} from '../package.json';
+import {AccessKeyConfigJson} from './server_access_key';
 
 import {ServerConfigJson} from './server_config';
 
@@ -64,6 +65,7 @@ export interface DailyFeatureMetricsReportJson {
 // Field renames will break backwards-compatibility.
 export interface DailyDataLimitMetricsReportJson {
   enabled: boolean;
+  perKeyLimitCount?: number;
 }
 
 export interface SharedMetricsPublisher {
@@ -151,11 +153,13 @@ export class OutlineSharedMetricsPublisher implements SharedMetricsPublisher {
   private reportStartTimestampMs: number;
 
   // serverConfig: where the enabled/disable setting is persisted
+  // keyConfig: where access keys are persisted
   // usageMetrics: where we get the metrics from
   // toMetricsId: maps Access key ids to metric ids
   // metricsUrl: where to post the metrics
   constructor(
       private clock: Clock, private serverConfig: JsonConfig<ServerConfigJson>,
+      private keyConfig: JsonConfig<AccessKeyConfigJson>,
       usageMetrics: UsageMetrics,
       private toMetricsId: (accessKeyId: AccessKeyId) => AccessKeyMetricsId,
       private metricsCollector: MetricsCollectorClient) {
@@ -233,11 +237,15 @@ export class OutlineSharedMetricsPublisher implements SharedMetricsPublisher {
   }
 
   private async reportFeatureMetrics(): Promise<void> {
+    const keys = this.keyConfig.data().accessKeys;
     const featureMetricsReport = {
       serverId: this.serverConfig.data().serverId,
       serverVersion: version,
       timestampUtcMs: this.clock.now(),
-      dataLimit: {enabled: !!this.serverConfig.data().accessKeyDataLimit},
+      dataLimit: {
+        enabled: !!this.serverConfig.data().accessKeyDataLimit,
+        perKeyLimitCount: keys.filter(key => !!key.dataLimit).length
+      }
     };
     await this.metricsCollector.collectFeatureMetrics(featureMetricsReport);
   }
@@ -250,4 +258,4 @@ function hasSanctionedCountry(countries: string[]) {
     }
   }
   return false;
-}
\ No newline at end of file
+}