Kafka Classification Filtering

These procedures will test classification access controls for Kafka.

1. Requirements

Data Fabric is operating at a classification level of SECRET (or greater).
Access to Keycloak admin console.
kcat installed locally.

2. Setup

2.1. Ensure Keycloak Group membership Mapper Enabled

In Keycloak, under the data-fabric realm,

Select the Client scopes menu.
Select profile scope.
Click the Mappers tab.
Click the Add mapper → By configuration button.
Select Group Membership.
Input:
- Name: groups
- Token Claim Name: groups
click Save.

If you get an error stating:

Could not create mapping: 'Protocol mapper exists with same name'

then the mapper is already configured.

2.2. Create Test Users

As you create each user, save the user’s ID (a uuid) to a file or somewhere you can easily copy it. You will use it later in these procedures.

For the purposes of this test, we will create 3 users:

test-u - Test user with UNCLASSIFIED clearance, USA citizen.
test-s - Test user with SECRET clearance, USA citizen.
test-s-gbr - Test user with SECRET clearance, GBR citizen.

All data used in the test is synthetic and will be deleted at the end of the test to avoid confusion.

Login to the Keycloak admin console at https://DF_HOST/auth.
Navigate to realm data-fabric → Users.
Click the Add user button.
- Username: test-u
Click Create
Under the Attributes tab, click Add an attribute to enter each of the following key/value pairs:
- country: USA
- classification: UNCLASSIFIED
Click Save.
Under the Credentials tab, click Set password and input:
- Password: test
- Password confirmation: test
- Temporary: toggle to "Off"
Click Save → Save password (confirmation).

Repeat the steps above for user test-s, with the only difference being:

classification to SECRET.

Repeat the steps above for user test-s-gbr, with the only difference being:

country: GBR
classification: SECRET

2.3. Create Test Topics

Login to Data Fabric at https://DF_HOST as the test-s user.
From the nav menu, click APIs to open the Swagger docs.
In the root Swagger listing under Fabric Services, click the /api/v1/kafka link.
In Swagger, verify the test-s user is the one currently authenticated in the upper right Authorize button .

If a different user is authorized, logout and enter the test-s / test credentials into the df_basic authorization and click the Authorize button.
Under the Topics section expand the POST /topics row and click the Try it out button.

2.3.1. Create Topic: `test-topic-u`

For the Parameters, input:
- classification: U

In the Request body section set the "name" to "test-topic-u", input:

{
  "name": "test-topic-u",
  "partitions": 1,
  "replicas": 1
}

Click the Execute button.
Verify you get back a 200 response with a JSON payload indicating the topic was created successfully.

2.3.2. Create Topic: `test-topic-s`

For the Parameters, input:
- classification: S//NF

In the Request body section set the "name" to "test-topic-s", input:

{
  "name": "test-topic-s",
  "partitions": 1,
  "replicas": 1
}

Click the Execute button.
Verify you get back a 200 response with a JSON payload indicating the topic was created successfully.

2.3.3. Create Topic: `test-topic-s-fvey`

For the Parameters, input:
- classification: S//REL TO USA, FVEY

In the Request body section set the "name" to "test-topic-s-fvey", input:

{
  "name": "test-topic-s-fvey",
  "partitions": 1,
  "replicas": 1
}

Click the Execute button.
Verify you get back a 200 response with a JSON payload indicating the topic was created successfully.

3. Basic Tests

3.1. List Topics for User: `test-u`

This will demonstrate that the test-u user cannot see the existence of topics that exceed his clearance.

In Swagger, authenticate yourself as user test-u / test (using df_basic authorization).
Expand the GET /topics row and click the Try it out button.
Click the Execute button.
Verify you get back a 200 response with a list of topics that includes:
- test-topic-u
  
  but does NOT include:
- test-topic-s
- test-topic-s-fvey

3.2. List Topics for User: `test-s`

This will demonstrate that the test-s user can access SECRET topics based on their clearance.

In Swagger, authenticate yourself as user test-s / test (using df_basic authorization).
Expand the GET /topics row and click the Try it out button.
Click the Execute button.
Verify you get back a 200 response with a list of topics that includes:
- test-topic-u
- test-topic-s
- test-topic-s-fvey

3.3. List Topics for User: `test-s-gbr`

This will demonstrate that the test-s-gbr user can access FVEY topics, but not those restricted by NF.

In Swagger, authenticate yourself as user test-s-gbr / test (using df_basic authorization).
Expand the GET /topics row and click the Try it out button.
Click the Execute button.
Verify you get back a 200 response with a list of topics that includes:
- test-topic-s-fvey
  
  but does NOT include:
- test-topic-s

4. Client Tests

These tests will demonstrate that users can connect directly to Kafka, but only to topic(s) they have the clearance to access.

4.1. Unauthorized Clients Cannot Access Kafka

Only users who have been authorized by an administrator can access Kafka.

In Swagger, authenticate yourself as user test-u / test (using df_basic authorization).
Expand the GET /client row and click the Try it out button.
Click the Execute button.
Verify you get back a 401 error response with a reason header stating that the user has not been authorized for client access.
```
reason: User '...' has not been authorized for client access.
```

4.2. Authorize Clients for Kafka Access

Administrators can grant users access to Kafka.

In Swagger, authenticate yourself as user admin (using df_basic authorization).
Expand the PUT /users/{user} row and click the Try it out button.
For the Parameters, input:
- user: <the principal ID for the test-u user> (from Keycloak when you created the user above).
Click the Execute button.
Verify you get back a 200 response with a payload echoing the user’s principal you just granted. Similar to:
```
{
  "name": "ebda6af6-828e-4b3f-88fa-3be9c955cd2d"
}
```

Repeat the steps above for users test-s and test-s-gbr.

4.3. Connect as User: `test-u`

In Swagger, authenticate yourself as user test-u / test (using df_basic authorization).
Expand the GET /client row and click the Try it out button.
Click the Execute button.
Verify you get back a 200 response with a application/zip payload.
In the response details, click the Download file link to download the df-kafka.zip file.
Extract df-kafka.zip and verify it contains:
- df-kafka.crt - a certificate for the client’s TLS connection.
- kcat.conf - config file for a kcat client to connect to the cluster.
View the contents of kcat.conf, and copy the example usage from comment at the top of the file.
From a shell in the same directory that you extracted the zip, execute the example usage but change the topic to test-topic-u.
```
kcat -b kafka-bootstrap.DF_HOST:443 -C -t test-topic-u -F kcat.conf
```
The example usage in kcat.conf will have the actual host in place of DF_HOST.

Verify your client successfully subscribes to the test-topic-u topic, which should look like:

% Reading configuration from file kcat.conf
% Reached end of topic test-topic-u [0] at offset 0

Kill the kcat client session.

Attempt to subscribe to topic to test-topic-s.

kcat -b kafka-bootstrap.DF_HOST:443 -C -t test-topic-s -F kcat.conf

Verify the test-u user is DENIED access to topic test-topic-s with an error stating:

% Reading configuration from file kcat.conf
% ERROR: Topic test-topic-s [0] error: Fetch from broker 2 failed: Broker: Topic authorization failed

Kill the kcat client session.
Delete the df-kafka.zip and contents that were extracted from it.

4.4. Connect as User: `test-s`

In Swagger, authenticate yourself as user test-s / test (using df_basic authorization).
Expand the GET /client row and click the Try it out button.
Click the Execute button.
Verify you get back a 200 response with a application/zip payload.
In the response details, click the Download file link to download the df-kafka.zip file.
Extract df-kafka.zip and verify it contains:
- df-kafka.crt - a certificate for the client’s TLS connection.
- kcat.conf - config file for a kcat client to connect to the cluster.
View the contents of kcat.conf, and copy the example usage from comment at the top of the file.
From a shell in the same directory that you extracted the zip, execute the example usage but change the topic to test-topic-s.
```
kcat -b kafka-bootstrap.DF_HOST:443 -C -t test-topic-s -F kcat.conf
```
The example usage in kcat.conf will have the actual host in place of DF_HOST.

Verify your client successfully subscribes to the test-topic-s topic, which should look like:

% Reading configuration from file kcat.conf
% Reached end of topic test-topic-s [0] at offset 0

Kill the kcat client session.
Delete the df-kafka.zip and contents that were extracted from it.

4.5. Connect as User: `test-s-gbr`

In Swagger, authenticate yourself as user test-s-gbr / test (using df_basic authorization).
Expand the GET /client row and click the Try it out button.
Click the Execute button.
Verify you get back a 200 response with a application/zip payload.
In the response details, click the Download file link to download the df-kafka.zip file.
Extract df-kafka.zip and verify it contains:
- df-kafka.crt - a certificate for the client’s TLS connection.
- kcat.conf - config file for a kcat client to connect to the cluster.
View the contents of kcat.conf, and copy the example usage from comment at the top of the file.
From a shell in the same directory that you extracted the zip, execute the example usage but change the topic to test-topic-s-fvey.
```
kcat -b kafka-bootstrap.DF_HOST:443 -C -t test-topic-s-fvey -F kcat.conf
```
The example usage in kcat.conf will have the actual host in place of DF_HOST.

Verify your client successfully subscribes to the test-topic-s-fvey topic, which should look like:

% Reading configuration from file kcat.conf
% Reached end of topic test-topic-s-fvey [0] at offset 0

Kill the kcat client session.

Attempt to subscribe to topic to test-topic-s.

kcat -b kafka-bootstrap.DF_HOST:443 -C -t test-topic-s -F kcat.conf

Verify the test-s-gbr user is DENIED access to topic test-topic-s with an error stating:

% Reading configuration from file kcat.conf
% ERROR: Topic test-topic-s [0] error: Fetch from broker 2 failed: Broker: Topic authorization failed

Kill the kcat client session.
Delete the df-kafka.zip and contents that were extracted from it.

5. Topic Deletion Tests

These tests will demonstrate that only administrator users can delete topics.

5.1. Non-Admins Cannot Delete Topics

In Swagger, authenticate yourself as user test-u / test (using df_basic authorization).
Expand the DELETE /topics/{topic} row and click the Try it out button.
For the Parameters, input:
- topic: test-topic-u
Click the Execute button.
Verify you get back a 403 response with a header www-authenticate that indicates the user has insufficient privileges.

5.2. Admins Can Delete Topics

In Swagger, authenticate yourself as user admin (using df_basic authorization).
Expand the DELETE /topics/{topic} row and click the Try it out button.
For the Parameters, input:
- topic: test-topic-u
Click the Execute button.
Verify you get back a 200 response with a payload containing the name of the deleted topic.
```
{
  "name": "test-topic-u"
}
```

Repeat the steps above for topics test-topic-s and test-topic-s-fvey to clean up.