API Security, Authentication and Authorization Questions

Direct answer

Test field-level authorization the same way you would test any access-control rule: run the
identical query as two different callers (a caller who should see a protected field and one who
should not) and assert on the difference. A GraphQL server that enforces field rules only in the
resolver, and not also in the schema's introspection output, will happily tell an unauthorized
caller a sensitive field exists even while refusing to return its value, so the test also needs
to work when introspection is turned off in an environment.

Structured elaboration

Why this is a distinct risk from REST-style object-level authorization. In a REST API you
typically authorize per endpoint or per object id. In GraphQL, a single query can request many
fields on an object in one call, and each field can have its own authorization rule (this is the
GraphQL-specific instance of what OWASP (the Open Web Application Security Project, a nonprofit that publishes ranked lists of common API and web vulnerability categories) calls, in its 2023 API Security Top 10, Broken Object Property
Level Authorization, API3:2023). It is easy to correctly gate the object as a whole (only the
owner can query user(id)) while forgetting that one of its fields, say ssn or
internalRiskScore, should be admin-only regardless of who owns the parent object.

Building the test:

1. Enumerate the fields to check. In an environment where introspection is enabled, query
   the schema itself to get the full field list for the type under test. In production-like
   environments introspection is usually disabled as a hardening step, so the test needs a
   fallback: a checked-in SDL (schema definition language) snapshot of the type, kept current by
   a CI check that diffs it against the schema whenever the schema changes. Do not let "no
   introspection here" become "no test coverage here."
2. Run the query as the high-privilege caller and record which fields came back populated.
   This is your baseline of what the type is supposed to expose to someone with full access.
3. Run the identical query shape as the low-privilege caller.
4. Assert on the difference, not just on errors. Many GraphQL servers do not reject the
   whole request when a caller cannot read one field; they return null for that field plus an
   entry in the response's errors array (a "partial success"). A test that only checks the top
   level HTTP status code, or only checks for the presence of an error, will miss a server that
   silently returns the real value instead of null. Assert directly that every field which
   should be role-gated is either absent, null, or explicitly redacted in the low-privilege
   response, and separately assert that fields which should be visible did not regress.

Worked example

# Field-level authorization test for a GraphQL API.
#
# In real CI this would POST to a live endpoint (e.g. with `requests` or `gql`),
# sending the query with an admin token, then again with a low-privilege token.
# To keep this example runnable with no network dependency, the "server" below
# is a tiny in-process resolver enforcing the same rule a real GraphQL server
# would via a per-field `@auth(role: ...)` directive: it nulls out any field
# the caller's role does not satisfy. The test logic (enumerate -> re-run as
# low-priv -> assert unauthorized fields absent) is exactly what you would run
# against a live endpoint.

FIELD_ROLES = {
    "id": None,
    "email": None,
    "fullName": None,
    "ssn": "ADMIN",
    "internalRiskScore": "ADMIN",
    "billingAddress": "ADMIN",
}

USER_RECORD = {
    "id": "u-42",
    "email": "jane@example.com",
    "fullName": "Jane Doe",
    "ssn": "123-45-6789",
    "internalRiskScore": 87,
    "billingAddress": "1 Market St",
}

def execute_query(fields, role):
    # Returns the same shape a real GraphQL response would: unauthorized
    # fields come back null with an entry in `errors`, matching how most
    # GraphQL servers signal a partial-authorization failure.
    data, errors = {}, []
    for f in fields:
        required = FIELD_ROLES[f]
        if required is None or required == role:
            data[f] = USER_RECORD[f]
        else:
            data[f] = None
            errors.append(f"not authorized to read field '{f}'")
    return {"data": data, "errors": errors}

def schema_fields_for_type(type_name, introspection_enabled=True):
    # Falls back to a checked-in SDL snapshot when introspection is disabled
    # (a common prod hardening step), instead of depending on a live
    # introspection query that only works in dev/staging.
    return list(FIELD_ROLES.keys())

def test_field_level_authorization_bypass():
    all_fields = schema_fields_for_type("User", introspection_enabled=True)

    # 1) enumerate what an admin can see
    admin_response = execute_query(all_fields, role="ADMIN")
    admin_visible = {f for f, v in admin_response["data"].items() if v is not None}
    assert admin_visible == set(all_fields)

    # 2) repeat the identical query shape for a low-privilege caller
    user_response = execute_query(all_fields, role="USER")
    user_visible = {f for f, v in user_response["data"].items() if v is not None}

    # 3) assert every admin-only field is absent (redacted to null) for the low-priv caller
    admin_only_fields = {f for f, req in FIELD_ROLES.items() if req == "ADMIN"}
    leaked = admin_only_fields & user_visible
    assert not leaked, f"low-privilege caller could read admin-only fields: {leaked}"

    public_fields = set(all_fields) - admin_only_fields
    assert public_fields <= user_visible

    return {
        "admin_visible": sorted(admin_visible),
        "user_visible": sorted(user_visible),
        "admin_only_fields_blocked_for_user": sorted(admin_only_fields),
        "user_errors": user_response["errors"],
    }

result = test_field_level_authorization_bypass()
print("admin_visible:", result["admin_visible"])
print("user_visible:", result["user_visible"])
print("admin_only_fields_blocked_for_user:", result["admin_only_fields_blocked_for_user"])
print("user_errors:", result["user_errors"])
print("PASS: no admin-only field leaked to the low-privilege caller")

Output:

admin_visible: ['billingAddress', 'email', 'fullName', 'id', 'internalRiskScore', 'ssn']
user_visible: ['email', 'fullName', 'id']
admin_only_fields_blocked_for_user: ['billingAddress', 'internalRiskScore', 'ssn']
user_errors: ["not authorized to read field 'ssn'", "not authorized to read field 'internalRiskScore'", "not authorized to read field 'billingAddress'"]
PASS: no admin-only field leaked to the low-privilege caller

The admin caller sees all 6 fields; the low-privilege caller sees only the 3 public fields, and
the 3 admin-only fields (ssn, internalRiskScore, billingAddress) come back null with a
matching error. A bug that forgot to gate one of those fields would flip that field from null
into user_visible, and the assertion on leaked would fail immediately, which is the exact
regression this test is designed to catch.

Trade-offs and pitfalls

Handling schema introspection differences between environments. Many teams disable
introspection in production for defense-in-depth, but leave it on in staging or dev. If your
test only discovers the field list by querying introspection live, it silently stops running in
the one environment where you most need it. Keep a versioned SDL snapshot as the source of truth
for "what fields exist," and add a separate, lightweight CI check that fails whenever a live
introspection query (in an environment where it is enabled) diverges from the snapshot, so the
snapshot itself cannot go stale.

Pitfall: testing only the object, not its fields. A team that already has a BOLA (Broken Object Level Authorization, OWASP API1:2023, a different category from the object-property-level one described earlier in this answer) style test
("user A cannot fetch user B's record") sometimes assumes field-level authorization is covered
by the same test. It is not: BOLA is about whose object you can reach, field-level
authorization is about which parts of an object you can see once you can reach it, and a
caller can legitimately own the object (their own profile) while still not being entitled to
every field on it (an internal risk score, say). Test the two separately.

Pitfall: asserting on HTTP status only. GraphQL servers conventionally return HTTP 200 even
for partially-authorized or partially-failed responses, with the real signal inside the response
body's data and errors. A test suite carried over from REST-style thinking that only checks
response.status_code == 200 will pass even when a field-level authorization bug leaks real
data, because the status code never changes.

Direct answer

A streaming endpoint (WebSocket or Server-Sent Events, SSE) needs its own test design because
the security properties that matter for a request/response API, a valid handshake, a bounded
request rate, are joined here by properties that only exist because the connection is
long-lived: message-level integrity over time, protection against replaying an old message into
a live session, and resource exhaustion from connections rather than just requests. Build the
test suite around a small client harness that can open many concurrent connections, drive each
one through a scripted message sequence, and assert on both the wire-level responses and the
server's resource behavior under that load.

Structured elaboration

Authentication handshake. A WebSocket upgrade or an SSE connection is still established over
plain HTTP first, so the same bearer token or session cookie can (and should) be validated
during that initial handshake, before the connection is upgraded. Test that: a request with no
credential, an expired token, and a token for the wrong scope are all rejected at the handshake,
not silently accepted and only checked later on the first message. Also test that a token which
expires while the connection is open is handled deliberately (either the connection is closed
when the token expires, or there is an explicit re-authentication message on the wire); a
long-lived connection that never re-checks the token turns a short-lived access token into a
de facto long-lived one.

Message integrity. For SSE (server to client only), integrity mostly means the server should
sign or otherwise make tamperable proxies detectable if messages traverse an untrusted
intermediary. For WebSocket (bidirectional), also verify the server rejects malformed frames and
does not trust client-supplied fields (a userId or role embedded in a message body) over the
identity established at the handshake.

Replay protection. Because the connection is long-lived, a captured message can potentially
be replayed within the same session or against a new one. Test with a monotonically increasing
sequence number or nonce per message: send a valid message, then resend the identical message
(or an old captured one) and assert the server rejects or ignores the duplicate rather than
re-applying its effect (for example, re-processing a trade order or a chat message twice).

Payload schema validation. Every inbound client message should be validated against a schema
just like a REST request body would be, since a long-lived socket does not get the benefit of a
fresh request going through a gateway's schema layer on every message; test with malformed JSON,
missing required fields, and fields of the wrong type.

Rate limiting and denial-of-service protection. Test at three levels: per-message rate within
a single connection (a client sending messages far faster than any legitimate client would),
per-connection limits (maximum connections from one identity or IP), and total connection-count
exhaustion (opening many connections and never sending anything, to see if idle connections are
reaped).

Simulating many clients. Use an async client library that can hold many concurrent
connections cheaply in one process (Python's websockets or asyncio with aiohttp, or a
purpose-built load tool like k6 with its WebSocket support, or artillery); a
thread-per-connection design will run out of threads long before it stresses the server. Drive
each simulated client through the same scripted sequence (connect, authenticate, send N
messages, replay one, disconnect) and collect per-connection results centrally rather than
asserting inside each client, so one flaky connection does not hide the rest of the results.

Worked example

Concretely, a CI job for this might run three staged checks against a small local instance of
the streaming service, each producing a pass/fail count rather than one giant assertion:

1. Handshake matrix (5 cases: no token, expired token, wrong-scope token, valid token, valid
   token that expires 2 seconds into the test): assert the connection is accepted or rejected as
   expected, and for the "expires mid-connection" case, assert the server closes the connection
   (or sends an explicit re-auth challenge) within a bounded time of expiry rather than leaving
   it open indefinitely.
2. Replay check: open one authenticated connection, send message seq=1 (a value-changing
   action), assert the expected effect happened once, resend the identical seq=1 message, and
   assert the effect did not happen a second time.
3. Load and DoS check: open 500 simulated connections concurrently, each sending messages at
   twice the documented per-connection rate limit; assert that the server's rejection rate
   converges toward "every message past the limit is rejected" rather than the server accepting
   an unbounded backlog, and assert the server's own health/liveness endpoint stays responsive
   throughout, since a common failure mode is the abuse traffic itself starving the server's
   ability to answer anything, including its own health check.

Below is the structural shape of that harness (illustrative, not executed here, since it depends
on a running server and a WebSocket library that is not part of this answer):

async def scripted_client(url, token, messages, results):
    async with connect(url, headers={"Authorization": f"Bearer {token}"}) as ws:
        for msg in messages:
            await ws.send(json.dumps(msg))
        responses = [json.loads(m) async for m in ws]
        results.append(responses)

async def run_load_check(url, token, n_clients, messages_per_client):
    results = []
    await asyncio.gather(*[
        scripted_client(url, token, build_messages(messages_per_client), results)
        for _ in range(n_clients)
    ])
    return results

Trade-offs and pitfalls

• SSE and WebSocket are not symmetric. SSE is server-to-client only, so "message integrity"
  and "replay protection" apply to what the server emits, and the client has no message channel
  to abuse; do not write a client-message replay test against an SSE endpoint, since there is no
  such message to replay. Confirm which protocol is in play before reusing a test suite between
  them.
• A pure black-box load test can miss server-side resource leaks. High concurrent connection
  counts that pass a black-box assertion (connections were accepted, messages were rejected past
  the limit) can still be leaking memory or file descriptors per connection; pair the black-box
  test with a server-side resource check (open file descriptors, memory) taken before and after
  the load run.
• Common wrong turn: testing the handshake once and assuming the connection stays governed by
  it. The whole point of a long-lived connection is that authorization state can drift after
  the handshake (permissions revoked, token expired, user disabled); a test suite that never
  probes mid-connection state changes will miss that class of bug entirely.
• Common wrong turn: single-client functional tests standing in for load tests. A test that
  proves one client's auth, integrity, and replay checks pass says nothing about what happens
  when 10,000 clients do the same thing at once; keep the functional correctness tests and the
  concurrent load tests as separate suites so a slow load test does not block fast feedback on
  the correctness checks.

Direct answer

A timing side-channel test compares the distribution of response times for two request
classes (valid username vs. invalid username), not a single measurement, because one request
is dominated by network jitter. Build it as: collect many interleaved paired samples, trim
outliers, run a distribution-comparison statistic that does not assume clean symmetric noise,
and require the result to repeat across several independent CI runs before failing a build, so
one noisy run cannot trip a false positive.

Structured elaboration

Why the leak exists. An endpoint that does extra work only when a username is real (looks
up a stored password hash, runs a bcrypt/argon2 compare) systematically takes longer for valid
usernames, even when the HTTP response body and status code look identical. That gap lets an
attacker enumerate real usernames purely from timing, with no error message needed.

Reduce noise before comparing anything:

• Warm up the service and discard the first N requests (cold caches, JIT/interpreter warmup,
  TCP slow start all inflate early samples).
• Interleave the two request types (alternate valid, invalid, valid, invalid...) instead of
  running one whole batch then the other. Load drift, garbage collection pauses, and neighbor
  noisy-tenant effects then hit both groups equally instead of biasing one.
• Trim outliers (for example the top and bottom 5%) before computing a summary statistic.
  Network jitter and retransmits produce occasional huge spikes that would otherwise swamp a
  genuine sub-millisecond gap.

Choosing a sample size. The bigger the real timing gap relative to the noise, the fewer
samples you need to detect it reliably. A gap of a few milliseconds against light noise might
separate cleanly with a few dozen samples; a sub-millisecond gap buried in several milliseconds
of network jitter (the realistic case for a well-behaved API) typically needs several hundred
to a few thousand samples per group. Start around 300 to 500 per group, and increase it if
independent runs disagree with each other.

Statistical analysis. Do not compare raw means with a standard two-sample t-test (a statistic that compares two groups' averages, assuming the noise around each is roughly symmetric and bell-curve-shaped): network
latency is right-skewed (a long tail of slow outliers, not a symmetric bell curve), which
violates the t-test's assumptions and makes it unreliable here. A permutation test on the
trimmed data is a better fit: it makes no assumption about the shape of the noise, works
directly on the trimmed samples, and gives an empirical p-value (the probability of seeing a gap at least this large purely by chance, if there were no real timing difference) by asking "how often would a
gap this large appear if the labels were random?"

Automate in CI without false positives from network variance:

• Fix an explicit significance threshold (a low one, since you will run the test often) and
  require several independent runs to agree before failing the build. A single p < 0.05 result
  will happen by chance on noisy timing data with some regularity.
• Run on a dedicated or otherwise quiet CI runner where possible, so background load is not
  itself a confound.
• Include a negative control in the same pipeline: the same statistical test comparing
  valid-vs-valid samples should almost never fire. If it does, your pipeline's noise floor is
  too high to trust the real test's result that day.

Worked example

The script below simulates response times for 400 requests per group: a shared 12ms base cost,
a real 0.6ms leak injected only into the "invalid username" branch (modeling an extra hashed
comparison), and network jitter modeled as a right-skewed exponential distribution (mean 3ms)
rather than a symmetric one, since that is closer to real network noise. It trims the top and
bottom 5% of each group, then runs a one-sided permutation test. It checks 5 independent
"CI runs" (5 different random seeds) and only flags a leak if at least 4 of 5 agree at
p < 0.01, then repeats the whole pipeline with no injected leak as a negative control.

import random
import statistics

random.seed(42)

N = 400  # samples per group

def sample_response_times(base_ms, leak_ms, n):
    # base_ms: shared processing cost. leak_ms: the side-channel gap this
    # branch adds (0 for the "valid username" baseline). Jitter is modeled
    # as a right-skewed exponential, closer to real network noise than Gaussian.
    out = []
    for _ in range(n):
        jitter = random.expovariate(1 / 3.0)   # mean 3ms right-skewed jitter
        core = random.gauss(base_ms, 0.4)      # small stable variance in app code
        out.append(core + leak_ms + jitter)
    return out

def trim(samples, pct=0.05):
    # drop the top/bottom pct of samples to blunt network-spike outliers
    s = sorted(samples)
    k = int(len(s) * pct)
    return s[k: len(s) - k] if k > 0 else s

def permutation_test(a, b, n_perm=2000, seed=0):
    # one-sided permutation test for mean(b) > mean(a)
    rng = random.Random(seed)
    observed = statistics.mean(b) - statistics.mean(a)
    pooled = a + b
    na = len(a)
    count_ge = 0
    for _ in range(n_perm):
        rng.shuffle(pooled)
        perm_a = pooled[:na]
        perm_b = pooled[na:]
        diff = statistics.mean(perm_b) - statistics.mean(perm_a)
        if diff >= observed:
            count_ge += 1
    p_value = (count_ge + 1) / (n_perm + 1)  # avoids an impossible p=0
    return observed, p_value

def run_one_ci_check(leak_ms, run_seed):
    random.seed(run_seed)
    valid = sample_response_times(base_ms=12.0, leak_ms=0.0, n=N)
    invalid = sample_response_times(base_ms=12.0, leak_ms=leak_ms, n=N)
    observed_ms, p = permutation_test(trim(valid), trim(invalid), n_perm=2000, seed=run_seed)
    return observed_ms, p

# a real 0.6ms leak (e.g. an extra bcrypt compare only run when the username
# exists) buried under ~3ms mean network jitter, checked across 5 independent
# CI runs (5 seeds) instead of trusting a single run
alpha = 0.01
significant_runs = 0
print("run  observed_diff_ms  p_value  significant(p<0.01)")
for i, seed in enumerate([1, 2, 3, 4, 5]):
    observed_ms, p = run_one_ci_check(leak_ms=0.6, run_seed=seed)
    sig = p < alpha
    significant_runs += int(sig)
    print(f"{i+1:>3}  {observed_ms:>16.3f}  {p:>7.4f}  {sig}")
print(f"\nsignificant in {significant_runs}/5 runs (require >=4/5 to flag in CI)")
print("FLAG TIMING LEAK" if significant_runs >= 4 else "no consistent leak detected")

# negative control: no injected leak, same pipeline, to show the gate does
# not fire on jitter alone
print("\nnegative control (leak_ms=0.0):")
significant_runs_ctrl = 0
for i, seed in enumerate([11, 12, 13, 14, 15]):
    observed_ms, p = run_one_ci_check(leak_ms=0.0, run_seed=seed)
    sig = p < alpha
    significant_runs_ctrl += int(sig)
    print(f"{i+1:>3}  {observed_ms:>16.3f}  {p:>7.4f}  {sig}")
print(f"significant in {significant_runs_ctrl}/5 runs (require >=4/5 to flag in CI)")
print("FLAG TIMING LEAK" if significant_runs_ctrl >= 4 else "no consistent leak detected (correct: no leak was injected)")

Output:

run  observed_diff_ms  p_value  significant(p<0.01)
  1             0.433   0.0045  True
  2             0.696   0.0005  True
  3             0.627   0.0005  True
  4             0.771   0.0005  True
  5             0.318   0.0170  False

significant in 4/5 runs (require >=4/5 to flag in CI)
FLAG TIMING LEAK

negative control (leak_ms=0.0):
  1            -0.014   0.5402  False
  2            -0.215   0.8966  False
  3            -0.068   0.6727  False
  4            -0.070   0.6737  False
  5            -0.051   0.6207  False
significant in 0/5 runs (require >=4/5 to flag in CI)
no consistent leak detected (correct: no leak was injected)

With a real 0.6ms leak, 4 of 5 simulated CI runs cross the p < 0.01 threshold, so the gate
correctly fires. With no injected leak, all 5 runs come back non-significant, so the negative
control confirms the pipeline does not fire on jitter alone. Run 5 in the first block (p =
0.017) illustrates exactly why a single-run threshold is unsafe: that one run alone would have
been called "not significant" at the 0.01 level even though the leak was real, which is why the
gate requires 4 of 5 runs to agree rather than trusting any single run.

Trade-offs and pitfalls

• Statistical detection is not the same as measuring real-world exploitability. This test
  tells you a leak likely exists locally; it does not tell you how many requests an attacker
  over the public internet, with far more noise than your CI network, would need to exploit it.
  Those are two different measurements, and a leak that is statistically detectable in a quiet
  CI environment may be much harder to exploit remotely.
• Common wrong turn: comparing raw means on unfiltered data with a standard t-test. Network
  jitter is heavy-tailed, so a handful of retransmits can swing a raw mean far more than a
  genuine sub-millisecond leak, producing both false positives and false negatives.
• Common wrong turn: trusting a single p-value. Noisy timing measurements will occasionally
  produce a "significant" result by chance; requiring several independent runs to agree, as in
  the worked example, controls that risk in a way a single run cannot.
• Pitfall: "fixing" the endpoint's happy path but not its neighbors. If you make the
  invalid-username branch add a matching artificial delay but a cache layer, a database index,
  or connection pooling upstream still behaves differently for existing vs. non-existing users,
  the leak often survives at a smaller magnitude. Re-run the same statistical test as a
  regression gate after any fix, not just once at discovery time.