Merge pull request #286 from OzJacobii/feature/Version_Bump_5_0_14

5.0.14

axonius_api_client/api/asset_callbacks/base.py

@@ -183,6 +183,7 @@ def args_map_base(cls) -> dict:
             "field_null_value": None,
             "field_null_value_complex": [],
             "tags_add": [],
+            "expirable_tags": [],
             "tags_add_invert_selection": False,
             "tags_remove": [],
             "tags_remove_invert_selection": False,
@@ -805,6 +806,7 @@ def do_tagging(self):
     def do_tag_add(self):
         """Add tags to assets."""
         tags = listify(self.get_arg_value("tags_add"))
+        expirable_tags = self.get_arg_value("expirable_tags")
         rows = self.TAG_ROWS_ADD
         invert_selection = self.get_arg_value("tags_add_invert_selection")
         count_tags = len(tags)
@@ -817,7 +819,7 @@ def do_tag_add(self):
         if tags:
             self.echo(["Performing API call to add tags to assets", *msgs])
             count_modified = self.APIOBJ.labels.add(
-                rows=rows, labels=tags, invert_selection=invert_selection
+                rows=rows, labels=tags, invert_selection=invert_selection, expirable_tags=expirable_tags,
             )
             self.echo(msg=[f"API added tags to {count_modified} assets", *msgs])
 
@@ -1449,6 +1451,7 @@ def arg_export_fd_close(self) -> bool:
     "field_null_value": "Null value to use for missing simple fields",
     "field_null_value_complex": "Null value to use for missing complex fields",
     "tags_add": "Tags to add to assets",
+    "expirable_tags": "Expiration dates for tags",
     "tags_add_invert_selection": "Invert selection for tags to add",
     "tags_remove": "Tags to remove from assets",
     "tags_remove_invert_selection": "Invert selection for tags to remove",

axonius_api_client/api/assets/labels.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 """API for working with tags for assets."""
-from typing import List, Union
+from datetime import datetime, timedelta
+from typing import List, Optional, Union
 
 from ...tools import listify
 from .. import json_api
@@ -34,7 +35,7 @@ def get(self) -> List[str]:
             >>> apiobj: axonapi.api.assets.AssetMixin = client.devices
             >>>       # or client.users or client.vulnerabilities
             >>> apiobj.labels.get()
-            ['tag1', 'tag2']
+            ['tag1', 'tag2', 'tag3']
 
         """
         return [x.value for x in self._get()]
@@ -56,8 +57,46 @@ def get_expirable_names(self) -> List[str]:
         """
         return [x.value for x in self._get_expirable_names()]
 
+    @staticmethod
+    def _set_expirable_tags(expirations: dict) -> List[dict]:
+        """
+            Get dict with tags as keys and expiration date as values
+            expiration date can be as either
+             - date as a string (YYYY-MM-DD)
+             - int specify the "days from now"
+            Converts it to a List of dicts for each tag
+        Args:
+            expirations: Dict with tag name and tag expiration date
+            {'tag1': '2024-01-01', 'tag2': 5}
+        Returns:
+            List of dicts, each dict contains single tag name and tag expiration date
+            [{'name': 'tag1', 'expiration_date': '2024-01-01'}, ...]
+        """
+        expirable_tags: list = []
+        if not expirations:
+            return expirable_tags
+
+        for tag_name, tag_expiration_date in expirations.items():
+            if isinstance(tag_expiration_date, int) or tag_expiration_date.isdigit():
+                tag_expiration_date = int(tag_expiration_date)
+                tag_expiration_date = str(datetime.today() + timedelta(days=tag_expiration_date))
+            elif isinstance(tag_expiration_date, str):
+                # If the format is not correct, an exception will be raised
+                datetime.strptime(tag_expiration_date, '%Y-%m-%d')
+
+            expirable_tags.append({
+                'name': tag_name,
+                'expiration_date': tag_expiration_date,
+            })
+
+        return expirable_tags
+
     def add(
-        self, rows: Union[List[dict], str], labels: List[str], invert_selection: bool = False
+        self,
+        rows: Union[List[dict], str],
+        labels: List[str],
+        invert_selection: bool = False,
+        expirable_tags: Optional[dict] = None,
     ) -> int:
         """Add tags to assets.
 
@@ -69,25 +108,33 @@ def add(
             >>> client: axonapi.Connect = axonapi.Connect(**connect_args)
             >>> apiobj: axonapi.api.assets.AssetMixin = client.devices
             >>>       # or client.users or client.vulnerabilities
-            >>> data = apiobj.get(wiz_entries=[{'type': 'simple', 'value': 'name equals test'}])
-            >>> len(data)
+            >>> rows = apiobj.get(wiz_entries=[{'type': 'simple', 'value': 'name equals test'}])
+            >>> len(rows)
             1
 
-            >>> apiobj.labels.add(rows=rows, labels=['api tag 1', 'api tag 2'])
+            >>> apiobj.labels.add(
+                    rows=rows,
+                    labels=['api tag 1', 'api tag 2'],
+                    expirable_tags={'api tag 1': '2024-01-25', 'api tag 2': 5},
+                )
             1
 
         Args:
             rows: list of internal_axon_id strs or list of assets returned from a get method
             labels: tags to add
             invert_selection: True=add tags to assets that ARE NOT supplied in rows;
                 False=add tags to assets that ARE supplied in rows
+            expirable_tags: Dict with tag name and expiration_date (string or int) as keys
+             - expiration_date as string is a date (YYYY-MM-DD)
+             - expiration_date as int is days from now
 
         """
         ids = self._get_ids(rows=rows)
-        return self._add(labels=labels, ids=ids, include=not invert_selection).value
+        expirable_tags: List[dict] = self._set_expirable_tags(expirations=expirable_tags)
+        return self._add(labels=labels, ids=ids, include=not invert_selection, expirable_tags=expirable_tags).value
 
     def _add(
-        self, labels: List[str], ids: List[str], include: bool = True
+        self, labels: List[str], ids: List[str], include: bool = True, expirable_tags: List[dict] = None,
     ) -> json_api.generic.IntValue:
         """Direct API method to add labels/tags to assets.
 
@@ -96,11 +143,14 @@ def _add(
             ids: internal_axon_id of assets to add tags to
             include: True=add tags to assets that ARE supplied in rows;
                 False=add tags to assets that ARE NOT supplied in rows
+            expirable_tags: List of dicts, each dict with tag name and expiration_date
         """
         api_endpoint = ApiEndpoints.assets.tags_add
 
         entities = {"ids": listify(ids), "include": include}
-        request_obj = api_endpoint.load_request(entities=entities, labels=listify(labels))
+        request_obj = api_endpoint.load_request(
+            entities=entities, labels=listify(labels), expirable_tags=expirable_tags,
+        )
         return api_endpoint.perform_request(
             http=self.auth.http, request_obj=request_obj, asset_type=self.asset_type
         )

axonius_api_client/api/json_api/assets/modify_tags_request.py

@@ -29,6 +29,13 @@ class ModifyTagsRequestSchema(BaseSchemaJson):
         allow_none=True,
         description="Filter to use to select assets?",
     )
+    expirable_tags = mm_fields.List(
+        mm_fields.Dict(),
+        load_default=list,
+        dump_default=list,
+        allow_none=True,
+        description="List of dict with tags and expiration dates",
+    )
 
     class Meta:
         """JSONAPI config."""
@@ -51,6 +58,7 @@ class ModifyTagsRequest(BaseModel):
     entities: dict = field_from_mm(SCHEMA, "entities")
     labels: t.List[str] = field_from_mm(SCHEMA, "labels")
     filter: t.Optional[str] = field_from_mm(SCHEMA, "filter")
+    expirable_tags: t.Optional[list] = field_from_mm(SCHEMA, "expirable_tags")
 
     SCHEMA: t.ClassVar[t.Any] = SCHEMA
 

axonius_api_client/cli/grp_assets/grp_common.py

@@ -169,6 +169,34 @@ def wiz_callback(ctx, param, value):
     return "\n".join(contents)
 
 
+def expirable_tags_callback(ctx, param, values) -> dict:
+    """
+    Args:
+        values: set of values, each value is a string with the format of tag_name=expiration_date
+                  - each value must contain "=" sign
+                expiration date can be either
+                  - date as a string (YYYY-MM-DD)
+                  - int specify the "days from now"
+                example: ('E=2024-05-05', 'F=10')
+    Returns:
+        Dict with tags as keys and expiration date as values
+        example: {'E': '2024-05-05', 'F': '10'}
+    """
+    expirations: dict = {}
+
+    if not values:
+        return []
+
+    for value in values:
+        if "=" not in value:
+            echo_error(msg=f"Invalid expression, must contain tag_name=expired_date or tag_name=days_from_now")
+
+        tag_name, tag_expiration_date = value.split('=')
+        expirations[tag_name] = tag_expiration_date
+
+    return expirations
+
+
 WIZ = [
     click.option(
         "--wiz",
@@ -471,6 +499,17 @@ def wiz_callback(ctx, param, value):
         hidden=False,
         metavar="TAG",
     ),
+    click.option(
+        "--expirable_tags",
+        "-exp",
+        "expirable_tags",
+        help="Expirable date for tag to set in the format of tag_name=YYYY-MM-DD or tag_name=days_from_now (multiple)",
+        multiple=True,
+        show_envvar=True,
+        show_default=True,
+        required=False,
+        callback=expirable_tags_callback,
+    ),
     click.option(
         "--tag-invert/--no-tag-invert",
         "tags_add_invert_selection",

axonius_api_client/version.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 """Version information for this package."""
 
-__version__ = "5.0.13"
+__version__ = "5.0.14"
 VERSION: str = __version__
 """Version of package."""
 
@@ -37,6 +37,6 @@
 LICENSE: str = __license__
 """License of package."""
 
-__copyright__ = "Copyright Axonius 2023"
+__copyright__ = "Copyright Axonius 2024"
 COPYRIGHT: str = __copyright__
 """Copyright of package."""