diff --git a/splunklib/client.py b/splunklib/client.py index 038980ea4..5184698d0 100644 --- a/splunklib/client.py +++ b/splunklib/client.py @@ -64,6 +64,7 @@ import socket from datetime import datetime, timedelta from time import sleep +from typing import Any from urllib import parse try: @@ -109,6 +110,7 @@ def deprecated(message): # pyright: ignore[reportUnknownParameterType] PATH_DEPLOYMENT_SERVERCLASSES = "deployment/serverclass/" PATH_EVENT_TYPES = "saved/eventtypes/" PATH_FIRED_ALERTS = "alerts/fired_alerts/" +PATH_HEC_TOKENS = "data/inputs/http/" PATH_INDEXES = "data/indexes/" PATH_INPUTS = "data/inputs/" PATH_JOBS = "search/jobs/" @@ -498,6 +500,14 @@ def fired_alerts(self): """ return Collection(self, PATH_FIRED_ALERTS, item=AlertGroup) + @property + def hec_tokens(self): + """Returns the collection of HTTP Event Collector tokens. + + :return: A :class:`HECTokens` collection of :class:`HECToken` entities. + """ + return HECTokens(self) + @property def indexes(self): """Returns the collection of indexes for this Splunk instance. @@ -2126,6 +2136,39 @@ def count(self): return int(self.content.get("triggered_alert_count", 0)) +class HECToken(Entity): + """This class represents an HTTP Event Collector token.""" + + def __init__(self, service: "Service", path: str, **kwargs: Any) -> None: + Entity.__init__(self, service, path, **kwargs) + + def disable(self) -> "HECToken": # pyright: ignore[reportImplicitOverride] + """Disables this HEC token. + + :return: The :class:`HECToken`. + """ + self.post(disabled="1") + self.refresh() + return self + + def enable(self) -> "HECToken": # pyright: ignore[reportImplicitOverride] + """Enables this HEC token. + + :return: The :class:`HECToken`. + """ + self.post(disabled="0") + self.refresh() + return self + + +class HECTokens(Collection): + """This class represents a collection of HTTP Event Collector tokens. + Retrieve this collection using :meth:`Service.hec_tokens`.""" + + def __init__(self, service: "Service") -> None: + Collection.__init__(self, service, PATH_HEC_TOKENS, item=HECToken) + + class Indexes(Collection): """This class contains the collection of indexes in this Splunk instance. Retrieve this collection using :meth:`Service.indexes`. diff --git a/tests/unit/test_hec_token.py b/tests/unit/test_hec_token.py new file mode 100644 index 000000000..8d73b2150 --- /dev/null +++ b/tests/unit/test_hec_token.py @@ -0,0 +1,34 @@ +# Copyright © 2011-2026 Splunk, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"): you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from splunklib.client import ( + PATH_HEC_TOKENS, + Collection, + Entity, + HECToken, + HECTokens, +) + + +class TestHECToken: + def test_is_entity_subclass(self) -> None: + assert issubclass(HECToken, Entity) + + def test_path_constant(self) -> None: + assert PATH_HEC_TOKENS == "data/inputs/http/" + + +class TestHECTokens: + def test_is_collection_subclass(self) -> None: + assert issubclass(HECTokens, Collection)