Add v2 endpoint for telephony log (#208)

* Add v2 endpoint for telephony log

* Fix test by correcting expected output

Author: AaronAtDuo

.gitignore

@@ -9,3 +9,6 @@ dist
 .idea
 env/
 py3env/
+.venv
+duo_client.egg-info
+*.DS_Store

duo_client/admin.py

@@ -174,44 +174,39 @@
 import six.moves.urllib
 
 from . import client
+from .logs.telephony import Telephony
 import six
 import warnings
 import time
 import base64
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
-USER_STATUS_ACTIVE = 'active'
-USER_STATUS_BYPASS = 'bypass'
-USER_STATUS_DISABLED = 'disabled'
-USER_STATUS_LOCKED_OUT = 'locked out'
+USER_STATUS_ACTIVE = "active"
+USER_STATUS_BYPASS = "bypass"
+USER_STATUS_DISABLED = "disabled"
+USER_STATUS_LOCKED_OUT = "locked out"
 
-TOKEN_HOTP_6 = 'h6'
-TOKEN_HOTP_8 = 'h8'
-TOKEN_YUBIKEY = 'yk'
+TOKEN_HOTP_6 = "h6"
+TOKEN_HOTP_8 = "h8"
+TOKEN_YUBIKEY = "yk"
 
 VALID_AUTHLOG_REQUEST_PARAMS = [
-    'mintime',
-    'maxtime',
-    'limit',
-    'sort',
-    'next_offset',
-    'event_types',
-    'reasons',
-    'results',
-    'users',
-    'applications',
-    'groups',
-    'factors',
-    'api_version'
+    "mintime",
+    "maxtime",
+    "limit",
+    "sort",
+    "next_offset",
+    "event_types",
+    "reasons",
+    "results",
+    "users",
+    "applications",
+    "groups",
+    "factors",
+    "api_version",
 ]
 
-VALID_ACTIVITY_REQUEST_PARAMS = [
-    'mintime',
-    'maxtime',
-    'limit',
-    'sort',
-    'next_offset'
-]
+VALID_ACTIVITY_REQUEST_PARAMS = ["mintime", "maxtime", "limit", "sort", "next_offset"]
 
 
 class Admin(client.Client):
@@ -598,12 +593,12 @@ def get_activity_logs(self, **kwargs):
                         "value" : <int: total objects in the time range>
                     }
                 }
-            },
+            }
 
         Raises RuntimeError on error.
         """
         params = {}
-        today = datetime.utcnow()
+        today = datetime.now(tz=timezone.utc)
         default_maxtime = int(today.timestamp() * 1000)
         default_mintime = int((today - timedelta(days=180)).timestamp() * 1000)
 
@@ -622,8 +617,6 @@ def get_activity_logs(self, **kwargs):
         if 'limit' in params:
             params['limit'] = str(int(params['limit']))
 
-
-
         response = self.json_api_call(
             'GET',
             '/admin/v2/logs/activity',
@@ -634,41 +627,64 @@ def get_activity_logs(self, **kwargs):
             row['host'] = self.host
         return response
 
-    def get_telephony_log(self,
-                          mintime=0):
+    def get_telephony_log(self, mintime=0, api_version=1, **kwargs):
         """
         Returns telephony log events.
 
         mintime - Fetch events only >= mintime (to avoid duplicate
-                  records that have already been fetched)
-
-        Returns:
+            records that have already been fetched)
+        api_version - The API version of the handler to use.
+            Currently, the default api version is v1, but the v1 API
+            will be deprecated in a future version of the Duo Admin API.
+            Please migrate to the v2 api at your earliest convenience.
+            For details on the differences between v1 and v2,
+            please see Duo's Admin API documentation. (Optional)
+
+        v1 Returns:
             [
-                {'timestamp': <int:unix timestamp>,
-                 'eventtype': "telephony",
-                 'host': <str:host>,
-                 'context': <str:context>,
-                 'type': <str:type>,
-                 'phone': <str:phone number>,
-                 'credits': <str:credits>}, ...
+                {
+                    'timestamp': <int:unix timestamp>,
+                    'eventtype': "telephony",
+                    'host': <str:host>,
+                    'context': <str:context>,
+                    'type': <str:type>,
+                    'phone': <str:phone number>,
+                    'credits': <str:credits>}
             ]
+            
+        v2 Returns:
+            {
+                "items": [
+                    {
+                        'context': <str>,
+                        'credits': <int: credits used>,
+                        'phone': <str:phone number>,
+                        'telephony_id': <str:UUID>,
+                        'ts': <str:ISO timestamp>,
+                        'txid': <str:UUID>,
+                        'type': <str:"sms" or "phone">,
+                        'eventtype': <str:"telephony">,
+                        'host': <str:application hostname>
+                    }
+                ],
+                "metadata": {
+                    "next_offset": <str: comma seperated ts and offset value>
+                    "total_objects": {
+                        "relation" : <str: relational operator>
+                        "value" : <int: total objects in the time range>
+                    }
+                }
+            }
 
         Raises RuntimeError on error.
         """
-        # Sanity check mintime as unix timestamp, then transform to string
-        mintime = str(int(mintime))
-        params = {
-            'mintime': mintime,
-        }
-        response = self.json_api_call(
-            'GET',
-            '/admin/v1/logs/telephony',
-            params,
-        )
-        for row in response:
-            row['eventtype'] = 'telephony'
-            row['host'] = self.host
-        return response
+
+        if api_version not in [1,2]:
+            raise ValueError("Invalid API Version")
+
+        if api_version == 2:
+            return Telephony.get_telephony_logs_v2(self.json_api_call, self.host, **kwargs)
+        return Telephony.get_telephony_logs_v1(self.json_api_call, self.host, mintime=mintime)
 
     def get_users_iterator(self):
         """

duo_client/logs/__init__.py

@@ -0,0 +1,6 @@
+from __future__ import absolute_import
+from .telephony import Telephony
+
+__all__ = [
+    'Telephony'
+]

duo_client/logs/telephony.py

@@ -0,0 +1,65 @@
+from typing import Callable
+from duo_client.util import (
+    get_params_from_kwargs,
+    get_log_uri,
+    get_default_request_times,
+)
+
+VALID_TELEPHONY_V2_REQUEST_PARAMS = [
+    "filters",
+    "mintime",
+    "maxtime",
+    "limit",
+    "sort",
+    "next_offset",
+    "account_id",
+]
+
+LOG_TYPE = "telephony"
+
+
+class Telephony:
+    @staticmethod
+    def get_telephony_logs_v1(json_api_call: Callable, host: str, mintime=0):
+        # Sanity check mintime as unix timestamp, then transform to string
+        mintime = f"{int(mintime)}"
+        params = {
+            "mintime": mintime,
+        }
+        response = json_api_call(
+            "GET",
+            get_log_uri(LOG_TYPE, 1),
+            params,
+        )
+        for row in response:
+            row["eventtype"] = LOG_TYPE
+            row["host"] = host
+        return response
+
+    @staticmethod
+    def get_telephony_logs_v2(json_api_call: Callable, host: str, **kwargs):
+        params = {}
+        default_mintime, default_maxtime = get_default_request_times()
+
+        params = get_params_from_kwargs(VALID_TELEPHONY_V2_REQUEST_PARAMS, **kwargs)
+
+        if "mintime" not in params:
+            # If mintime is not provided, the script defaults it to 180 days in past
+            params["mintime"] = default_mintime
+        params["mintime"] = f"{int(params['mintime'])}"
+        if "maxtime" not in params:
+            # if maxtime is not provided, the script defaults it to now
+            params["maxtime"] = default_maxtime
+        params["maxtime"] = f"{int(params['maxtime'])}"
+        if "limit" in params:
+            params["limit"] = f"{int(params['limit'])}"
+
+        response = json_api_call(
+            "GET",
+            get_log_uri(LOG_TYPE, 2),
+            params,
+        )
+        for row in response["items"]:
+            row["eventtype"] = LOG_TYPE
+            row["host"] = host
+        return response

duo_client/util.py

@@ -0,0 +1,21 @@
+from typing import Dict, Sequence, Tuple
+from datetime import datetime, timedelta, timezone
+
+
+def get_params_from_kwargs(valid_params: Sequence[str], **kwargs) -> Dict:
+    params = {}
+    for k in kwargs:
+        if kwargs[k] is not None and k in valid_params:
+            params[k] = kwargs[k]
+    return params
+
+
+def get_log_uri(log_type: str, version: int = 1) -> str:
+    return f"/admin/v{version}/logs/{log_type}"
+
+
+def get_default_request_times() -> Tuple[int, int]:
+    today = datetime.now(tz=timezone.utc)
+    mintime = int((today - timedelta(days=180)).timestamp() * 1000)
+    maxtime = int(today.timestamp() * 1000) - 120
+    return mintime, maxtime