Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions .pubnub.yml
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
name: python
version: 10.7.2
version: 10.8.0
schema: 1
scm: github.com/pubnub/python
sdks:
Expand All @@ -18,7 +18,7 @@ sdks:
distributions:
- distribution-type: library
distribution-repository: package
package-name: pubnub-10.7.2
package-name: pubnub-10.8.0
location: https://pypi.org/project/pubnub/
supported-platforms:
supported-operating-systems:
Expand Down Expand Up @@ -94,8 +94,8 @@ sdks:
-
distribution-type: library
distribution-repository: git release
package-name: pubnub-10.7.2
location: https://github.com/pubnub/python/releases/download/10.7.2/pubnub-10.7.2.tar.gz
package-name: pubnub-10.8.0
location: https://github.com/pubnub/python/releases/download/10.8.0/pubnub-10.8.0.tar.gz
supported-platforms:
supported-operating-systems:
Linux:
Expand Down Expand Up @@ -169,6 +169,11 @@ sdks:
license-url: https://github.com/encode/httpx/blob/master/LICENSE.md
is-required: Required
changelog:
- date: 2026-10-19
version: 10.8.0
changes:
- type: feature
text: "Add category-level permissions for App Context enumeration."
- date: 2026-07-08
version: 10.7.2
changes:
Expand Down
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,9 @@
## 10.8.0
October 19 2026

#### Added
- Add category-level permissions for App Context enumeration.

## 10.7.2
July 08 2026

Expand Down
40 changes: 40 additions & 0 deletions pubnub/endpoints/access/grant.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,15 @@

class Grant(Endpoint):
GRANT_PATH = "/v2/auth/grant/sub-key/%s"
SUPPORTED_CATEGORIES = ("channels", "uuids")

def __init__(self, pubnub):
Endpoint.__init__(self, pubnub)
self._auth_keys = []
self._channels = []
self._groups = []
self._uuids = []
self._categories = []
self._read = None
self._write = None
self._manage = None
Expand Down Expand Up @@ -54,6 +56,15 @@ def channel_groups(self, channel_groups):
utils.extend_list(self._groups, channel_groups)
return self

def categories(self, categories):
""" Grants category-level (enumeration) permissions on a whole App Context resource type.

Supported values are `channels` and `uuids`. Requires an auth key and allows
only the `get` permission; the server rejects anything else. Revoke with `get(False)`.
"""
utils.extend_list(self._categories, categories)
return self

def read(self, flag):
self._read = flag
return self
Expand Down Expand Up @@ -118,6 +129,9 @@ def custom_params(self):
if self._uuids:
params['target-uuid'] = utils.join_items(self._uuids)

if self._categories:
params['category'] = utils.join_items(self._categories)

Comment on lines +132 to +134

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

validate_params method doesn't have verification for the case when user specify categories we need to ensure that auth has been passed as well as required.
No check on what exactly has been passed for the categories and there is only channels and uuids allowed.
Also, if categories not empty we need to verify that only get permissions has been granted.

@jguz-pubnub jguz-pubnub Oct 12, 2026 •

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does the server already enforce these rules? I see the spec documents a 400 for each case with a distinct message. Anyway, I'll add these validations

if self._ttl is not None:
params['ttl'] = str(int(self._ttl))

Expand Down Expand Up @@ -149,6 +163,32 @@ def validate_params(self):
and self._update is None and self._join is None:
raise PubNubException(pn_error=PNERR_PAM_NO_FLAGS)

self.validate_categories()

def validate_categories(self):
if not self._categories:
return

invalid = [category for category in self._categories if category not in self.SUPPORTED_CATEGORIES]
if invalid:
raise PubNubException(
pn_error=PNERR_PAM_INVALID_ARGUMENTS,
errormsg="Unsupported categories: %s. Only 'channels' and 'uuids' are supported" % ", ".join(
str(category) for category in invalid))

if not self._auth_keys:
raise PubNubException(pn_error=PNERR_PAM_INVALID_ARGUMENTS,
errormsg="Category grants require providing non empty authKeys")

if any(flag is True for flag in (self._read, self._write, self._manage, self._delete, self._update,
self._join)):
raise PubNubException(pn_error=PNERR_PAM_INVALID_ARGUMENTS,
errormsg="Category grants allow only the 'get' permission")

if self._get is True and self._ttl is not None and int(self._ttl) == 1:
raise PubNubException(pn_error=PNERR_PAM_INVALID_ARGUMENTS,
errormsg="One-minute (ttl=1) category grants are not supported")

def create_response(self, envelope):
return PNAccessManagerGrantResult.from_json(envelope['payload'])

Expand Down
36 changes: 32 additions & 4 deletions pubnub/endpoints/access/grant_token.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
from typing import Union, List, Optional
from pubnub import utils
from pubnub.endpoints.endpoint import Endpoint
from pubnub.errors import PNERR_TTL_MISSING, PNERR_INVALID_META, PNERR_RESOURCES_MISSING
from pubnub.errors import PNERR_TTL_MISSING, PNERR_INVALID_META, PNERR_RESOURCES_MISSING, \
PNERR_PAM_INVALID_ARGUMENTS
from pubnub.exceptions import PubNubException
from pubnub.enums import HttpMethod, PNOperationType
from pubnub.enums import HttpMethod, PNOperationType, PAMPermissions
from pubnub.models.consumer.common import PNStatus
from pubnub.models.consumer.v3.access_manager import PNGrantTokenResult
from pubnub.models.consumer.v3.category import Category
from pubnub.models.consumer.v3.channel import Channel
from pubnub.models.consumer.v3.group import Group
from pubnub.models.consumer.v3.space import Space
Expand All @@ -24,7 +26,8 @@ class GrantToken(Endpoint):

def __init__(self, pubnub, channels: Union[str, List[str]] = None, channel_groups: Union[str, List[str]] = None,
users: Union[str, List[str]] = None, spaces: Union[str, List[str]] = None,
authorized_user_id: str = None, ttl: Optional[int] = None, meta: Optional[any] = None):
authorized_user_id: str = None, ttl: Optional[int] = None, meta: Optional[any] = None,
categories: Optional[List[Category]] = None):
Endpoint.__init__(self, pubnub)
self._ttl = ttl
self._meta = meta
Expand All @@ -42,6 +45,10 @@ def __init__(self, pubnub, channels: Union[str, List[str]] = None, channel_group
if users:
utils.extend_list(self._uuids, users)

self._categories = []
if categories:
utils.extend_list(self._categories, categories)

self._sort_params = True

def ttl(self, ttl: int) -> 'GrantToken':
Expand Down Expand Up @@ -80,6 +87,10 @@ def uuids(self, uuids: List[UUID]) -> 'GrantToken':
self._uuids = uuids
return self

def categories(self, categories: List[Category]) -> 'GrantToken':
self._categories = categories
return self

def custom_params(self):
return {}

Expand All @@ -99,6 +110,11 @@ def build_data(self):
permissions['resources'] = resources
permissions['patterns'] = patterns

if self._categories:
permissions['categories'] = {
category.get_id(): utils.calculate_bitmask(category) for category in self._categories
}

if self._meta:
if isinstance(self._meta, dict):
permissions['meta'] = self._meta
Expand All @@ -125,6 +141,7 @@ def validate_params(self):
self.validate_secret_key()
self.validate_ttl()
self.validate_resources()
self.validate_categories()

def create_response(self, envelope) -> PNGrantTokenResult:
return PNGrantTokenResult.from_json(envelope['data'])
Expand All @@ -148,9 +165,20 @@ def name(self):
return "Grant Token"

def validate_resources(self):
if not any((self._channels, self._groups, self._uuids)):
if not any((self._channels, self._groups, self._uuids, self._categories)):
raise PubNubException(pn_error=PNERR_RESOURCES_MISSING)

def validate_categories(self):
for category in self._categories:
if not isinstance(category, Category) or category.get_id() not in (Category.CHANNELS, Category.UUIDS):
raise PubNubException(
pn_error=PNERR_PAM_INVALID_ARGUMENTS,
errormsg="Unsupported category. Only Category.channels() and Category.uuids() are supported")

if utils.calculate_bitmask(category) != PAMPermissions.GET.value:
raise PubNubException(pn_error=PNERR_PAM_INVALID_ARGUMENTS,
errormsg="Category permissions must be exactly the 'get' permission")

def validate_ttl(self):
if not self._ttl:
raise PubNubException(pn_error=PNERR_TTL_MISSING)
9 changes: 9 additions & 0 deletions pubnub/managers.py
Original file line number Diff line number Diff line change
Expand Up @@ -431,6 +431,15 @@ def parse_token(cls, token):
token[resource_type][resource]
)

categories = {}
for wire_key, category_name in (("chan", "channels"), ("uuid", "uuids")):
permissions = token.get("cat", {}).get(wire_key)
if permissions is not None:
categories[category_name] = {"get": utils.has_get_permission(permissions)}

if categories:
parsed_token["categories"] = categories

return parsed_token

@staticmethod
Expand Down
14 changes: 13 additions & 1 deletion pubnub/models/consumer/access_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@


class _PAMResult:
def __init__(self, level, subscribe_key, channels, groups, uuids, ttl=None, r=None, w=None, m=None, d=None):
def __init__(self, level, subscribe_key, channels, groups, uuids, ttl=None, r=None, w=None, m=None, d=None,
categories=None):
self.level = level
self.subscribe_key = subscribe_key
self.channels = channels
Expand All @@ -15,12 +16,14 @@ def __init__(self, level, subscribe_key, channels, groups, uuids, ttl=None, r=No
self.write_enabled = w
self.manage_enabled = m
self.delete_enabled = d
self.categories = categories if categories is not None else {}

@classmethod
def from_json(cls, json_input):
constructed_channels = {}
constructed_groups = {}
constructed_uuids = {}
constructed_categories = {}

# only extract ttl, others are to be fetched on per uuid level
r, w, m, d, g, u, j, ttl = fetch_permissions(json_input)
Expand Down Expand Up @@ -72,6 +75,10 @@ def from_json(cls, json_input):
for uuid, value in json_input['uuids'].items():
constructed_uuids[uuid] = PNAccessManagerUuidsData.from_json(uuid, value)

if 'categories' in json_input:
for category_name, value in json_input['categories'].items():
constructed_categories[category_name] = PNAccessManagerCategoryData.from_json(category_name, value)

return cls(
level=json_input['level'],
subscribe_key=json_input['subscribe_key'],
Expand All @@ -83,6 +90,7 @@ def from_json(cls, json_input):
m=m,
d=d,
ttl=ttl,
categories=constructed_categories,
)


Expand Down Expand Up @@ -136,6 +144,10 @@ class PNAccessManagerUuidsData(_PAMEntityData):
pass


class PNAccessManagerCategoryData(_PAMEntityData):
pass


class PNAccessManagerKeyData(object):
def __init__(self, r, w, m, d, g, u, j, ttl=None):
self.read_enabled = r
Expand Down
23 changes: 23 additions & 0 deletions pubnub/models/consumer/v3/category.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
from pubnub.models.consumer.v3.pn_resource import PNResource


class Category(PNResource):
"""Category-level permission for a whole App Context resource type."""

CHANNELS = 'channels'
UUIDS = 'uuids'

def __init__(self, resource_name=None):
super(Category, self).__init__(resource_name)

@staticmethod
def channels():
return Category(resource_name=Category.CHANNELS)

@staticmethod
def uuids():
return Category(resource_name=Category.UUIDS)

def get(self):
self._get = True
return self
7 changes: 5 additions & 2 deletions pubnub/pubnub_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ def my_listener(message, event):
from pubnub.models.consumer.objects_v2.channel_members import PNUUID
from pubnub.models.consumer.objects_v2.common import MemberIncludes, MembershipIncludes
from pubnub.models.consumer.objects_v2.page import PNPage
from pubnub.models.consumer.v3.category import Category
from pubnub.models.subscription import PubNubChannel, PubNubChannelGroup, PubNubChannelMetadata, PubNubUserMetadata, \
PNSubscriptionRegistry, PubNubSubscriptionSet

Expand Down Expand Up @@ -571,7 +572,8 @@ def grant_token(
spaces: Union[str, List[str]] = None,
authorized_user_id: str = None,
ttl: Optional[int] = None,
meta: Optional[Any] = None
meta: Optional[Any] = None,
categories: Optional[List[Category]] = None
) -> GrantToken:
return GrantToken(
self,
Expand All @@ -581,7 +583,8 @@ def grant_token(
spaces=spaces,
authorized_user_id=authorized_user_id,
ttl=ttl,
meta=meta
meta=meta,
categories=categories
)

def revoke_token(self, token: str) -> RevokeToken:
Expand Down
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

setup(
name='pubnub',
version='10.7.2',
version='10.8.0',
description='PubNub Real-time push service in the cloud',
author='PubNub',
author_email='support@pubnub.com',
Expand Down
Loading
Loading