Security — SASL/SCRAM authentication & ACL authorization

4 h
requires KFK-02

From your world. SOX access controls: who may read/write which object, provable least privilege

Lock down a cluster with SASL/SCRAM + ACLs

The grader starts a cluster with an authorizer (StandardAuthorizer) and a SASL/SCRAM listener. Your job is to implement the functions that create credentials, grant a least-privilege ACL, and build the client config. Then an allowed principal can produce and a disallowed principal is denied. Edit secure.py.

Functions to implement

`scram_add_config(password) -> str` (pure)

Return the value that kafka-configs.sh expects after --add-config for a SCRAM-SHA-256 user, exactly: SCRAM-SHA-256=[password=<password>].

`create_scram_user(cluster, admin_bootstrap, username, password)`

Create the SCRAM credential in cluster metadata. Run kafka-configs.sh --alter --add-config <scram_add_config> --entity-type users --entity-name <username> with cluster.run_tool(...). Use admin_bootstrap (the trusted internal listener whose principal is a super user).

`grant_topic_write(cluster, admin_bootstrap, principal, topic)`

Grant produce-only rights (Write and Describe) on exactly topic to User:<principal>. Use kafka-acls.sh --add --allow-principal .... Grant nothing broader (least privilege).

`client_config(sasl_bootstrap, username, password) -> dict`

Return a producer config dict that authenticates over the SASL/SCRAM listener: bootstrap.servers, security.protocol="SASL_PLAINTEXT", sasl.mechanism="SCRAM-SHA-256", sasl.username, sasl.password.

How it is graded

Pure config (no broker): scram_add_config returns the exact string. client_config has the right SASL keys.
Live security (native broker + authorizer, no Docker): the grader creates app and evil. It grants app the produce ACL on a topic. Then it asserts these three points:
- app produces successfully.
- evil (no ACL) is denied with an authorization failure.
- The grant of app on topic A does not let it write topic B (the ACL is scoped).

Note: the lab uses SASL_PLAINTEXT to stay free of certificates. In production, you would use SASL_SSL so the channel is encrypted — see the lesson.

Starter code

secure.py

"""Starter — KFK-12. Implement the SASL/SCRAM + ACL functions. The stubs below
currently fail the tests (gate G1)."""

from __future__ import annotations

SCRAM_MECHANISM = "SCRAM-SHA-256"


def scram_add_config(password):
    """Return the --add-config value for a SCRAM user, e.g.
    'SCRAM-SHA-256=[password=secret]'."""
    # TODO
    return ""


def create_scram_user(cluster, admin_bootstrap, username, password):
    """Create a SCRAM credential via kafka-configs.sh (use cluster.run_tool)."""
    # TODO
    pass


def grant_topic_write(cluster, admin_bootstrap, principal, topic):
    """Grant Write + Describe on `topic` to User:<principal> via kafka-acls.sh."""
    # TODO
    pass


def client_config(sasl_bootstrap, username, password):
    """Return a producer config dict authenticating as `username` over SASL/SCRAM."""
    # TODO
    return {}

Hints

Hint 1

scram_add_config(password) returns the string kafka-configs.sh wants after --add-config: 'SCRAM-SHA-256=[password=<password>]'. create_scram_user runs kafka-configs.sh --alter --add-config <that> --entity-type users --entity-name <user> via cluster.run_tool.

Hint 2

grant_topic_write runs kafka-acls.sh --add --allow-principal User:<principal> --operation Write --operation Describe --topic <topic>. Use the PLAINTEXT bootstrap (the trusted internal listener whose principal is a super user) to manage ACLs.

Hint 3

client_config(sasl_bootstrap, user, password) returns the producer config dict: bootstrap.servers + security.protocol='SASL_PLAINTEXT' + sasl.mechanism='SCRAM-SHA-256' + sasl.username + sasl.password. The allowed principal must have the ACL; a principal with no ACL gets TOPIC_AUTHORIZATION_FAILED.

Run it locally

uv run de start KFK-12/01-lockdown
# edit workspace/KFK-12/01-lockdown/
uv run de test