// Module included in the following assemblies:
//
// assembly-securing-kafka-clients.adoc

[id='con-securing-client-authorization-{context}']
= User authorization

[role="_abstract"]
Use the `KafkaUser` custom resource to configure authorization rules for users (clients) that require access to a Kafka cluster. 
Configure the rules using the `authorization` property in `KafkaUser.spec`.
By specifying a `type`, you control what rules are used.

To use simple authorization, you set the `type` property to `simple` in `KafkaUser.spec.authorization`.
The simple authorization uses the Kafka Admin API to manage the ACL rules inside your Kafka cluster.
Whether ACL management in the User Operator is enabled or not depends on your authorization configuration in the Kafka cluster.

* For simple authorization, ACL management is always enabled.
* For OPA authorization, ACL management is always disabled.
  Authorization rules are configured in the OPA server.
* For Keycloak authorization, you can manage the ACL rules directly in Keycloak.
  You can also delegate authorization to the simple authorizer as a fallback option in the configuration.
  When delegation to the simple authorizer is enabled, the User Operator will enable management of ACL rules as well.
* For custom authorization using a custom authorization plugin, use the `supportsAdminApi` property in the `.spec.kafka.authorization` configuration of the `Kafka` custom resource to enable or disable the support.

Authorization is cluster-wide.
The authorization type must match the equivalent configuration in the `Kafka` custom resource.

If ACL management is not enabled, Strimzi rejects a resource if it contains any ACL rules.

If you're using a standalone deployment of the User Operator, ACL management is enabled by default.
You can disable it using the `STRIMZI_ACLS_ADMIN_API_SUPPORTED` environment variable.

If no authorization is specified, the User Operator does not provision any access rights for the user.
Whether such a `KafkaUser` can still access resources depends on the authorizer being used.
For example, for `simple` authorization, this is determined by the `allow.everyone.if.no.acl.found` configuration in the Kafka cluster.

== ACL rules

`simple` authorization uses ACL rules to manage access to Kafka brokers.

ACL rules grant access rights to the user, which you specify in the `acls` property.

For more information about the `AclRule` object, see the link:{BookURLConfiguring}#type-AclRule-reference[`AclRule` schema reference^].

== Super user access to Kafka brokers

If a user is added to a list of super users in a Kafka broker configuration,
the user is allowed unlimited access to the cluster regardless of any authorization constraints defined in ACLs in `KafkaUser`.

For more information on configuring super user access to brokers, see xref:con-securing-kafka-authorization-{context}[Kafka authorization].

== User quotas

You can configure the `spec` for the `KafkaUser` resource to enforce quotas so that a user does not exceed a configured level of access to Kafka brokers.
You can set size-based network usage and time-based CPU utilization thresholds.
You can also add a partition mutation quota to control the rate at which requests to change partitions are accepted for user requests.

.An example `KafkaUser` with user quotas
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  quotas:
    producerByteRate: 1048576 <1>
    consumerByteRate: 2097152 <2>
    requestPercentage: 55 <3>
    controllerMutationRate: 10 <4>
----
<1> Byte-per-second quota on the amount of data the user can push to a Kafka broker
<2> Byte-per-second quota on the amount of data the user can fetch from a Kafka broker
<3> CPU utilization limit as a percentage of time for a client group
<4> Number of concurrent partition creation and deletion operations (mutations) allowed per second

For more information on these properties, see the link:{BookURLConfiguring}#type-KafkaUserQuotas-reference[`KafkaUserQuotas` schema reference^].
