BED-10295: Model GitHub App registration ownership

descriptions/edges/GH_Owns.md

@@ -2,19 +2,28 @@
 
 ## General Information
 
-The traversable GH_Owns edge represents that an organization owns a repository. This edge establishes the foundation of the access control model by linking repositories to their owning organization. It is traversable because repository ownership is a critical relationship for understanding how organizational permissions cascade down to repository-level access, making it essential for attack path analysis.
+The traversable GH_Owns edge represents that a GitHub account owns an application or repository. For GitHub Apps, ownership identifies the account whose authorized administrators can manage the App registration and credentials. For repositories, ownership links the repository to its organization. These control relationships are traversable for attack-path analysis.
 
 ## Edge Schema
 
 | Source | Destination | Traversable |
 | --- | --- | --- |
+| `GH_Enterprise` | `GH_App` | `true` |
+| `GH_Organization` | `GH_App` | `true` |
 | `GH_Organization` | `GH_Repository` | `true` |
+| `GH_User` | `GH_App` | `true` |
 
 ## Diagram
 
 ```mermaid
 graph LR
-    n0["GH_Organization"]
-    n1["GH_Repository"]
+    n0["GH_Enterprise"]
+    n1["GH_App"]
+    n2["GH_Organization"]
+    n3["GH_Repository"]
+    n4["GH_User"]
     n0 -->|GH_Owns| n1
+    n2 -->|GH_Owns| n1
+    n2 -->|GH_Owns| n3
+    n4 -->|GH_Owns| n1
 ```

descriptions/nodes/GH_App.md

@@ -24,10 +24,12 @@ App definitions are retrieved via the public `GET /apps/{app_slug}` endpoint (no
 | `owner_login` | `string` | The login of the user or organization that owns the app. |
 | `owner_node_id` | `string` | The node_id of the user or organization that owns the app. |
 | `owner_type` | `string` | The type of the owner (e.g., `User`, `Organization`). |
+| `registration_origin` | `string` | Whether the app is tenant-owned, third-party, or unknown. |
 | `events` | `list[string]` | JSON string of the default webhook events the app subscribes to. |
 | `installations_count` | `integer` | The total number of installations of this app across all organizations. |
 | `created_at` | `datetime` | When the app was created. |
 | `updated_at` | `datetime` | When the app was last updated. |
+| `query_owner` | `string` | Query for the account that registered the app. |
 | `query_installations` | `string` | Query for installations. |
 
 ## Diagram
@@ -36,5 +38,11 @@ App definitions are retrieved via the public `GET /apps/{app_slug}` endpoint (no
 graph LR
     n0["GH_App"]
     n1["GH_AppInstallation"]
+    n2["GH_Enterprise"]
+    n3["GH_Organization"]
+    n4["GH_User"]
     n0 -->|GH_InstalledAs| n1
+    n2 -->|GH_Owns| n0
+    n3 -->|GH_Owns| n0
+    n4 -->|GH_Owns| n0
 ```

extension/schema.json

@@ -2121,7 +2121,7 @@
     },
     {
       "name": "GH_Owns",
-      "description": "Organization owns a repository",
+      "description": "GitHub account owns an application or repository",
       "is_traversable": true
     },
     {

src/openhound_github/lookup.py

@@ -118,6 +118,38 @@ def enterprise_organization_node_ids(self, enterprise_node_id: str):
             [enterprise_node_id],
         )
 
+    @lru_cache
+    def is_tenant_organization_node_id(self, node_id: str) -> bool:
+        """Return whether an organization belongs to the collected tenant."""
+        collected_org_id = self._find_single_object(
+            f"""SELECT node_id FROM {self.schema}.organizations WHERE node_id = ?""",
+            [node_id],
+        )
+        if collected_org_id is not None:
+            return True
+
+        enterprise_org_id = self._find_single_object(
+            f"""SELECT id FROM {self.schema}.enterprise_organizations WHERE id = ?""",
+            [node_id],
+        )
+        return enterprise_org_id is not None
+
+    @lru_cache
+    def is_collected_user_node_id(self, node_id: str) -> bool:
+        """Return whether a user node exists in collected organization or enterprise data."""
+        organization_user_id = self._find_single_object(
+            f"""SELECT id FROM {self.schema}.users WHERE id = ?""",
+            [node_id],
+        )
+        if organization_user_id is not None:
+            return True
+
+        enterprise_user_id = self._find_single_object(
+            f"""SELECT id FROM {self.schema}.enterprise_users WHERE id = ?""",
+            [node_id],
+        )
+        return enterprise_user_id is not None
+
     @lru_cache
     def _enterprise_runner_group_identity_for_inherited_org_group(
         self, org_node_id: str, group_name: str

src/openhound_github/models/app_installation.py

@@ -215,10 +215,12 @@ class GHAppProperties(GHNodeProperties):
         owner_login: The login of the user or organization that owns the app.
         owner_node_id: The node_id of the user or organization that owns the app.
         owner_type: The type of the owner (e.g., `User`, `Organization`).
+        registration_origin: Whether the app is tenant-owned, third-party, or unknown.
         events: JSON string of the default webhook events the app subscribes to.
         installations_count: The total number of installations of this app across all organizations.
         created_at: When the app was created.
         updated_at: When the app was last updated.
+        query_owner: Query for the account that registered the app.
         query_installations: Query for installations.
     """
 
@@ -231,10 +233,12 @@ class GHAppProperties(GHNodeProperties):
     owner_login: str | None = None
     owner_node_id: str | None = None
     owner_type: str | None = None
+    registration_origin: str | None = None
     events: list[str] | None = None
     installations_count: int | None = None
     created_at: datetime | None = None
     updated_at: datetime | None = None
+    query_owner: str | None = None
     query_installations: str | None = None
 
 
@@ -252,6 +256,29 @@ class Owner(BaseModel):
         icon="cube",
         properties=GHAppProperties,
     ),
+    edges=[
+        EdgeDef(
+            start=nk.ENTERPRISE,
+            end=nk.APP,
+            kind=ek.OWNS,
+            description="Enterprise owns GitHub App",
+            traversable=True,
+        ),
+        EdgeDef(
+            start=nk.ORGANIZATION,
+            end=nk.APP,
+            kind=ek.OWNS,
+            description="Organization owns GitHub App",
+            traversable=True,
+        ),
+        EdgeDef(
+            start=nk.USER,
+            end=nk.APP,
+            kind=ek.OWNS,
+            description="User owns GitHub App",
+            traversable=True,
+        ),
+    ],
 )
 class App(BaseAsset):
     """One record from `apps` → one GH_App node. Edges (GH_InstalledAs) are emitted by AppInstallationAsset."""
@@ -262,7 +289,7 @@ class App(BaseAsset):
     slug: str
     client_id: str | None = None
     node_id: str
-    owner: Owner
+    owner: Owner | None = None
     name: str
     description: str | None = None
     permissions: dict
@@ -282,6 +309,51 @@ def org_node_id(self) -> str | None:
             return self.collected_org_node_id
         return self._lookup.org_id_for_login(self.org_login)
 
+    @property
+    def registration_origin(self) -> str:
+        if self.owner is None:
+            return "unknown"
+
+        owner_type = self.owner.type.casefold()
+        if owner_type == "user":
+            return "third_party"
+
+        if owner_type == "organization":
+            if self._lookup.is_tenant_organization_node_id(self.owner.node_id):
+                return "tenant_owned"
+            return "third_party" if self._lookup.enterprise_id() else "unknown"
+
+        if owner_type == "enterprise":
+            enterprise_id = self._lookup.enterprise_id()
+            if enterprise_id is None:
+                return "unknown"
+            return (
+                "tenant_owned"
+                if self.owner.node_id == enterprise_id
+                else "third_party"
+            )
+
+        return "unknown"
+
+    @property
+    def registration_owner_node_id(self) -> str | None:
+        if self.owner is None:
+            return None
+
+        owner_type = self.owner.type.casefold()
+        if owner_type == "organization":
+            owner_exists = self._lookup.is_tenant_organization_node_id(
+                self.owner.node_id
+            )
+        elif owner_type == "enterprise":
+            owner_exists = self._lookup.enterprise_id() == self.owner.node_id
+        elif owner_type == "user":
+            owner_exists = self._lookup.is_collected_user_node_id(self.owner.node_id)
+        else:
+            owner_exists = False
+
+        return self.owner.node_id if owner_exists else None
+
     @property
     def as_node(self):
         aid = self.node_id
@@ -298,21 +370,34 @@ def as_node(self):
                 description=self.description,
                 external_url=self.external_url,
                 html_url=self.html_url,
-                # owner_login=self.owner_login,
-                # owner_node_id=self.owner_node_id,
-                # owner_type=self.owner_type,
+                owner_login=self.owner.login if self.owner else None,
+                owner_node_id=self.owner.node_id if self.owner else None,
+                owner_type=self.owner.type if self.owner else None,
+                registration_origin=self.registration_origin,
                 # permissions=self.permissions,
                 # installations_count=self.installations_count,
                 events=self.events,
                 created_at=self.created_at,
                 updated_at=self.updated_at,
+                query_owner=f"MATCH p=(owner)-[:GH_Owns]->(:GH_App {{node_id:'{aid}'}}) RETURN p",
                 query_installations=f"MATCH p=(:GH_App {{node_id:'{aid}'}})-[:GH_InstalledAs]->(:GH_AppInstallation) RETURN p",
             ),
         )
 
     @property
     def edges(self) -> list[Edge]:
-        return []
+        owner_node_id = self.registration_owner_node_id
+        if owner_node_id is None:
+            return []
+
+        return [
+            Edge(
+                kind=ek.OWNS,
+                start=EdgePath(value=owner_node_id, match_by="id"),
+                end=EdgePath(value=self.node_id, match_by="id"),
+                properties=EdgeProperties(traversable=True),
+            )
+        ]
 
 
 @app.asset(

tests/test_app_model.py

@@ -0,0 +1,121 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from openhound_github.models.app_installation import App, Owner
+
+
+def _app(owner: Owner | None) -> App:
+    return App(
+        id=123,
+        slug="example-app",
+        client_id="Iv1.example",
+        node_id="APP_NODE_123",
+        owner=owner,
+        name="Example App",
+        permissions={},
+        events=[],
+        org_login="installed-org",
+        org_node_id="ORG_NODE_789",
+    )
+
+
+def _owner(owner_type: str, node_id: str = "OWNER_NODE_456") -> Owner:
+    return Owner(
+        login="example-owner",
+        id=456,
+        node_id=node_id,
+        type=owner_type,
+    )
+
+
+def test_app_node_includes_registration_owner_properties() -> None:
+    app = _app(_owner("Organization"))
+    lookup = MagicMock()
+    lookup.is_tenant_organization_node_id.return_value = True
+    app._lookup = lookup
+
+    properties = app.as_node.properties
+
+    assert properties.owner_login == "example-owner"
+    assert properties.owner_node_id == "OWNER_NODE_456"
+    assert properties.owner_type == "Organization"
+    assert properties.query_owner == (
+        "MATCH p=(owner)-[:GH_Owns]->"
+        "(:GH_App {node_id:'APP_NODE_123'}) RETURN p"
+    )
+
+
+@pytest.mark.parametrize(
+    ("owner_type", "owner_node_id", "tenant_org", "enterprise_id", "expected"),
+    [
+        ("Organization", "ORG_NODE_1", True, "ENT_NODE_1", "tenant_owned"),
+        ("Organization", "EXTERNAL_ORG", False, "ENT_NODE_1", "third_party"),
+        ("Organization", "UNRESOLVED_ORG", False, None, "unknown"),
+        ("User", "USER_NODE_1", False, "ENT_NODE_1", "third_party"),
+        ("Enterprise", "ENT_NODE_1", False, "ENT_NODE_1", "tenant_owned"),
+        ("Enterprise", "EXTERNAL_ENT", False, "ENT_NODE_1", "third_party"),
+        ("Enterprise", "ENT_NODE_1", False, None, "unknown"),
+        ("Bot", "BOT_NODE_1", False, "ENT_NODE_1", "unknown"),
+    ],
+)
+def test_app_classifies_registration_origin(
+    owner_type: str,
+    owner_node_id: str,
+    tenant_org: bool,
+    enterprise_id: str | None,
+    expected: str,
+) -> None:
+    app = _app(_owner(owner_type, owner_node_id))
+    lookup = MagicMock()
+    lookup.is_tenant_organization_node_id.return_value = tenant_org
+    lookup.enterprise_id.return_value = enterprise_id
+    app._lookup = lookup
+
+    assert app.as_node.properties.registration_origin == expected
+
+
+def test_app_with_missing_owner_has_unknown_registration_origin() -> None:
+    app = _app(None)
+
+    properties = app.as_node.properties
+
+    assert properties.owner_login is None
+    assert properties.owner_node_id is None
+    assert properties.owner_type is None
+    assert properties.registration_origin == "unknown"
+
+
+@pytest.mark.parametrize(
+    ("owner_type", "owner_node_id"),
+    [
+        ("Organization", "ORG_NODE_1"),
+        ("Enterprise", "ENT_NODE_1"),
+        ("User", "USER_NODE_1"),
+    ],
+)
+def test_app_emits_registration_edge_for_collected_owner(
+    owner_type: str, owner_node_id: str
+) -> None:
+    app = _app(_owner(owner_type, owner_node_id))
+    lookup = MagicMock()
+    lookup.is_tenant_organization_node_id.return_value = True
+    lookup.enterprise_id.return_value = owner_node_id
+    lookup.is_collected_user_node_id.return_value = True
+    app._lookup = lookup
+
+    edge = app.edges[0]
+
+    assert edge.kind == "GH_Owns"
+    assert edge.start.value == owner_node_id
+    assert edge.end.value == "APP_NODE_123"
+    assert edge.properties.traversable is True
+
+
+def test_app_skips_registration_edge_for_uncollected_owner() -> None:
+    app = _app(_owner("Organization", "EXTERNAL_ORG"))
+    lookup = MagicMock()
+    lookup.is_tenant_organization_node_id.return_value = False
+    app._lookup = lookup
+
+    assert app.edges == []