1
import dns.message
2
import dns.asyncquery
3
import httpx
4

5
from ghunt.objects.base import GHuntCreds
6
from ghunt.apis.identitytoolkit import IdentityToolkitHttp
7

8

9
async def is_cloud_functions_panel_existing(project_id: str):
10
    q = dns.message.make_query(f"endpoints.{project_id}.cloud.goog", "A")
11
    r = await dns.asyncquery.tcp(q, "8.8.8.8")
12
    return bool(r.answer)
13

14
async def project_nb_from_key(as_client: httpx.AsyncClient, ghunt_creds: GHuntCreds, api_key: str, fallback=True) -> str|None:
15
    identitytoolkit_api = IdentityToolkitHttp(ghunt_creds)
16
    found, project_config = await identitytoolkit_api.get_project_config(as_client, api_key)
17
    if found:
18
        return project_config.project_id
19
    if fallback:
20
        # Fallback on fetching the project number by producing an error
21
        import json
22
        import re
23
        req = await as_client.get("https://blobcomments-pa.clients6.google.com/$discovery/rest", params={"key": api_key})
24
        try:
25
            data = json.loads(req.text)
26
            return re.findall(r'\d{12}', data["error"]["message"])[0]
27
        except Exception:
28
            pass
29
    return None
30