Merge pull request #10 from SpaceInvaderTech/experimental

migrate-part-of-script-to-aws

Author: chrisbek

.env.example

@@ -0,0 +1,4 @@
+PASSWD="extract this from iCloud keychain"
+HAYSTACKS_ENDPOINT=/haystacks
+USER_AGENT_COMMENT="Please contact [email protected]"
+API_KEY="get this from prod/space-invader-api/api-key"

.github/workflows/build-deploy.yml

@@ -0,0 +1,49 @@
+name: Deploy Serverless App
+
+on:
+  workflow_dispatch:
+    inputs:
+      stage:
+        description: 'Stage to deploy to'
+        required: false
+        default: 'prod'
+
+jobs:
+  deploy:
+    name: Deploy
+    runs-on: ubuntu-latest
+    # Add permissions block for OIDC token
+    permissions:
+      id-token: write
+      contents: read
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Configure AWS Credentials
+        uses: aws-actions/configure-aws-credentials@v2
+        with:
+          role-to-assume: arn:aws:iam::942778112339:role/git-deploy
+          aws-region: eu-central-1
+
+      - name: Install Node.js
+        uses: actions/setup-node@v3
+        with:
+          node-version: '18'
+
+      - name: Install npm dependencies
+        run: npm ci
+
+      - name: Install Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.11'
+
+      - name: Install Poetry
+        run: |
+          pipx install poetry==1.8.2
+          poetry install --no-root --no-interaction
+
+      - name: Deploy to AWS
+        run: ./node_modules/.bin/sls deploy --stage ${{ github.event.inputs.stage }}

.gitignore

@@ -164,3 +164,6 @@ status_code.txt
 log.txt
 
 test.sh
+.idea/
+node_modules/
+.serverless/

.python-version

@@ -1,27 +1,48 @@
 # AppleCollector
 
-Query Apple's Find My network, based on all the hard work of [OpenHaystack](https://github.com/seemoo-lab/openhaystack/), @vtky, @hatomist and others.
+Query Apple's Find My network, based on all the hard work
+of [OpenHaystack](https://github.com/seemoo-lab/openhaystack/), @vtky, @hatomist and others.
 
 This is a fork of great work from @biemster and modified to have device locations sent to an endpoint.
 
 ## Prerequisites
 
-XCode or Command Line Tools, latest pip (`pip3 install -U pip`, otherwise `cryptography` will not install).
+- Install python 3.12 and poetry on your system (`pipx install poetry==1.8.2`) if you don't have it already.
+- Install project dependencies: `poetry shell` and `poetry install`
 
-    pip3 install -r requirements.txt
+### Project Setup (MacOS)
 
-## Scripts
+- Enable iCloud on your macOS device
+- Search for `icloud` in the Keychain
+![img.png](docs/keychain_search.png)
 
-`main.py` will query Apple's Find My network based on private keys fetched from an API and can send locations to an API.
+- Select the `iCloud` entry with your email address
+![keychain_select.png](docs%2Fkeychain_select.png)
 
-`passwd.sh` will get a one time password for iCloud and store it in `$HOME/.haypass`.
+- Click on `show password` and copy the password
+![img.png](docs/keychain_show_pass.png)
 
-`example.cron.sh` is an example script for running `main.py`.
+- Use this password as your `PASSWD` in the `.env` file
 
-`launched.AppleCollector.plist` is for periodically running `cron.sh`.
 
-Setup:
+### Project Setup (non-MacOS)
 
-    mkdir -p ~/Library/LaunchAgents
-    cp launched.AppleCollector.plist ~/Library/LaunchAgents/
-    launchctl load -w ~/Library/LaunchAgents/launched.AppleCollector.plist
+You cannot use the `python manage.py refresh-credentials` command to refresh the credentials, since this require
+access to the macOS keychain (and it's depended on hardware).
+
+## Refresh Credentials
+
+- Use the `python manage.py refresh-credentials --schedule-location-fetching` 
+command to refresh the credentials (run on a MacOS, stored on AWS) and schedule location fetching (on AWS)
+
+
+- Use the `python manage.py refresh-credentials` 
+command to refresh the credentials (stored on AWS) without scheduling location fetching
+
+> This command is handy for testing purposes
+
+## Local Debug
+
+- After you have executed `python manage.py refresh-credentials` you have one minute before the credentials expire
+- Run `python manage.py fetch-locations --trackers E0D4FA128FA9,EC3987ECAA50,CDAA0CCF4128,EDDC7DA1A247,D173D540749D --limit 1000 --hours-ago 48`
+to fetch the locations of specific trackers

ReadMe.md

@@ -0,0 +1,40 @@
+import logging
+from requests import Session
+from app.dtos import DeviceResponse
+from app.helpers import status_code_success
+
+requestSession = Session()
+logger = logging.getLogger(__name__)
+
+
+def fetch_devices_metadata_from_space_invader_api(
+        url,
+        headers=None,
+        limit: int = 3000,
+        page: int = 0,
+) -> DeviceResponse:
+    response = requestSession.get(url, headers=headers, timeout=60, params={
+        "limit": limit,
+        "offset": page,
+    })
+    _handle_response(response)
+    return DeviceResponse(**response.json())
+
+
+def send_reports_to_api(url, data, headers=None):
+    """Send reports to API"""
+    if not url:
+        return
+
+    response = requestSession.post(
+        url,
+        headers=headers,
+        json=data,
+        timeout=60,
+    )
+    _handle_response(response)
+
+
+def _handle_response(response):
+    if not status_code_success(response.status_code):
+        raise Exception(f"Request failed with status code {response.status_code}: {response.text}")

api.py

@@ -0,0 +1,71 @@
+"""
+Fetch from Apple's acsnservice
+"""
+import logging
+from requests import Session
+
+from app.exceptions import AppleAuthCredentialsExpired
+from app.helpers import status_code_success
+from app.date import unix_epoch, date_milliseconds
+from pydantic import BaseModel, Field
+
+requestSession = Session()
+logger = logging.getLogger(__name__)
+
+
+class AppleLocation(BaseModel):
+    date_published: int = Field(alias="datePublished")
+    payload: str
+    description: str
+    id: str
+    status_code: int = Field(alias="statusCode")
+
+    class Config:
+        populate_by_name = True
+        validate_by_name = True
+
+
+class ResponseDto(BaseModel):
+    results: list[AppleLocation] = Field(default_factory=list)
+    statusCode: str
+    error: str = Field(default=None)
+
+    @property
+    def is_success(self) -> bool:
+        return self.statusCode == "200"
+
+
+def apple_fetch(security_headers: dict, ids, hours_ago: int = 1) -> ResponseDto:
+    logger.info("Fetching locations from Apple API for %s", ids)
+    startdate = unix_epoch() - hours_ago * 60 * 60
+    enddate = unix_epoch()
+
+    response = _acsnservice_fetch(security_headers, ids, startdate, enddate)
+
+    if not status_code_success(response.status_code):
+        if response.status_code == 401:
+            raise AppleAuthCredentialsExpired(response.reason)
+
+        logger.error('Error from Apple API: %s %s', response.status_code, response.reason)
+        return ResponseDto(error=response.reason, statusCode=str(response.status_code))
+
+    return ResponseDto(**response.json())
+
+
+def _acsnservice_fetch(security_headers, ids, startdate, enddate):
+    """Fetch from Apple's acsnservice"""
+    data = {
+        "search": [
+            {
+                "startDate": date_milliseconds(startdate),
+                "endDate": date_milliseconds(enddate),
+                "ids": ids,
+            }
+        ]
+    }
+    return requestSession.post(
+        "https://gateway.icloud.com/acsnservice/fetch",
+        headers=security_headers,
+        json=data,
+        timeout=60,
+    )

app/__init__.py

@@ -0,0 +1,49 @@
+import os
+import json
+import logging
+import functools
+
+from app.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+def api_auth_required(func):
+    """
+    Decorator to validate API authorization code in request headers.
+    Checks for x-api-key header and validates it against CREDENTIALS_API_KEY env variable.
+    """
+
+    @functools.wraps(func)
+    def wrapper(event, context):
+        headers = event.get('headers', {})
+        if not headers:
+            logger.warning("No headers found in request")
+            return {
+                "statusCode": 401,
+                "body": json.dumps({"error": "Unauthorized"})
+            }
+
+        api_code = None
+        for key, value in headers.items():
+            if key.lower() == 'x-api-key':
+                api_code = value
+                break
+
+        if not api_code:
+            logger.warning("No x-api-key header found in request")
+            return {
+                "statusCode": 401,
+                "body": json.dumps({"error": "Unauthorized"})
+            }
+
+        if api_code != settings.CREDENTIALS_API_KEY:
+            logger.warning("Invalid API code provided")
+            return {
+                "statusCode": 403,
+                "body": json.dumps({"error": "Forbidden"})
+            }
+
+        return func(event, context)
+
+    return wrapper