1
from ghunt.objects.base import GHuntCreds
2
from ghunt.errors import *
3
import ghunt.globals as gb
4
from ghunt.objects.apis import GAPI, EndpointConfig
5
from ghunt.parsers.drive import DriveCommentList, DriveFile, DriveChildList
6
from ghunt.knowledge import drive as drive_knowledge
7

8
import httpx
9

10
from typing import *
11
import inspect
12
import json
13

14

15
class DriveHttp(GAPI):
16
    def __init__(self, creds: GHuntCreds, headers: Dict[str, str] = {}):
17
        super().__init__()
18
        
19
        if not headers:
20
            headers = gb.config.headers
21

22
        base_headers = {}
23

24
        headers = {**headers, **base_headers}
25

26
        # Android OAuth fields
27
        self.api_name = "drive"
28
        self.package_name = "com.google.android.apps.docs"
29
        self.scopes = [
30
            "https://www.googleapis.com/auth/drive",
31
            "https://www.googleapis.com/auth/drive.file"
32
        ]
33

34
        self.hostname = "www.googleapis.com"
35
        self.scheme = "https"
36

37
        self._load_api(creds, headers)
38

39
    async def get_file(self, as_client: httpx.AsyncClient, file_id: str) -> Tuple[bool, DriveFile]:
40
        endpoint = EndpointConfig(
41
            name = inspect.currentframe().f_code.co_name,
42
            verb = "GET",
43
            data_type = None, # json, data or None
44
            authentication_mode = "oauth", # sapisidhash, cookies_only, oauth or None
45
            require_key = None, # key name, or None
46
        )
47
        self._load_endpoint(endpoint)
48

49
        base_url = f"/drive/v2internal/files/{file_id}"
50

51
        params = {
52
            "fields": ','.join(drive_knowledge.request_fields),
53
            "supportsAllDrives": True
54
        }
55

56
        req = await self._query(endpoint.name, as_client, base_url, params=params)
57

58
        # Parsing
59
        data = json.loads(req.text)
60
        drive_file = DriveFile()
61
        if "error" in data:
62
            return False, drive_file
63

64
        drive_file._scrape(data)
65

66
        return True, drive_file
67

68
    async def get_comments(self, as_client: httpx.AsyncClient, file_id: str, page_token: str="") -> Tuple[bool, str, DriveCommentList]:
69
        endpoint = EndpointConfig(
70
            name = inspect.currentframe().f_code.co_name,
71
            verb = "GET",
72
            data_type = None, # json, data or None
73
            authentication_mode = "oauth", # sapisidhash, cookies_only, oauth or None
74
            require_key = None, # key name, or None
75
        )
76
        self._load_endpoint(endpoint)
77

78
        base_url = f"/drive/v2internal/files/{file_id}/comments"
79

80
        params = {
81
            "supportsAllDrives": True,
82
            "maxResults": 100
83
        }
84

85
        if page_token:
86
            params["pageToken"] = page_token
87

88
        req = await self._query(endpoint.name, as_client, base_url, params=params)
89

90
        # Parsing
91
        data = json.loads(req.text)
92
        drive_comments = DriveCommentList()
93
        if "error" in data:
94
            return False, "", drive_comments
95

96
        next_page_token = data.get("nextPageToken", "")
97

98
        drive_comments._scrape(data)
99

100
        return True, next_page_token, drive_comments
101

102
    async def get_childs(self, as_client: httpx.AsyncClient, file_id: str, page_token: str="") -> Tuple[bool, str, DriveChildList]:
103
        endpoint = EndpointConfig(
104
            name = inspect.currentframe().f_code.co_name,
105
            verb = "GET",
106
            data_type = None, # json, data or None
107
            authentication_mode = "oauth", # sapisidhash, cookies_only, oauth or None
108
            require_key = None, # key name, or None
109
        )
110
        self._load_endpoint(endpoint)
111

112
        base_url = f"/drive/v2internal/files/{file_id}/children"
113

114
        params = {
115
            "supportsAllDrives": True,
116
            "maxResults": 1000
117
        }
118

119
        if page_token:
120
            params["pageToken"] = page_token
121

122
        req = await self._query(endpoint.name, as_client, base_url, params=params)
123

124
        # Parsing
125
        data = json.loads(req.text)
126
        drive_childs = DriveChildList()
127
        if "error" in data:
128
            return False, "", drive_childs
129

130
        next_page_token = data.get("nextPageToken", "")
131

132
        drive_childs._scrape(data)
133

134
        return True, next_page_token, drive_childs
135