inital commit

.github/workflows/deploy-pypi.yml

@@ -0,0 +1,54 @@
+name: Build pypi package
+
+on:
+  push:
+    tags: # on tags with versions
+      - "v*.*.*"
+
+jobs:
+  release-build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install poetry
+        run: pipx install poetry
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.x"
+          cache: "poetry"
+      - name: Install package and dependencies
+        run: |
+          poetry install --without dev --with build
+      - name: Build and publish
+        run: |
+          poetry version $(git describe --tags --abbrev=0)
+          poetry build
+      - name: Verify wheel using twine
+        run: |
+          poetry run twine check dist/* --strict
+      - name: Upload release distributions
+        uses: actions/upload-artifact@v4
+        with:
+          name: release-dists
+          path: dist/
+
+  pypi-publish:
+    runs-on: ubuntu-latest
+    needs:
+      - release-build
+    environment:
+      name: pypi
+      url: https://pypi.org/p/py-pdsadmin
+    permissions:
+      id-token: write
+    steps:
+      - name: Retrieve release distributions
+        uses: actions/download-artifact@v4
+        with:
+          name: release-dists
+          path: dist/
+      - name: Publish release distributions to PyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          print-hash: true

.gitignore

@@ -0,0 +1 @@
+__pycache__

LICENSE

@@ -0,0 +1,19 @@
+MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1 +1,44 @@
 # py-pdsadmin (Python PDS Admin)
+
+A Python implementation of [pdsadmin](https://github.com/bluesky-social/pds/tree/main/pdsadmin), installable with pipx.
+
+## Installation
+
+```bash
+pipx install py-pdsadmin
+```
+
+## Usage
+
+```bash
+pdsadmin list
+```
+
+```plaintext
+Available commands:
+  create-invite-code      Create an invite code for the PDS server.
+  help                    Displays help for a command.
+  list                    Lists commands.
+  request-crawl           Request a crawl of a PDS instance.
+
+ account
+  account create          Create a new account in the PDS server.
+  account delete          Delete an account from the PDS server.
+  account list            List accounts in the PDS server.
+  account reset-password  Reset the password of an account in the PDS server.
+  account takedown        Take down an account from the PDS server.
+  account untakedown      Untakedown an account in the PDS server.
+  account update-handle   Update the handle of an account in the PDS server.
+  ```
+
+### Notes on Usage
+
+Environment variables are automatically loaded from `/pds/pds.env` if it exists. This file is created by the [pds](https://github.com/bluesky-social/pds/tree/main) installation script. If you are not using the default installation, the script will look for `pds.env` or `.env` in the current working directory.
+
+## Contributing
+
+Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.
+
+## License
+
+[MIT](https://choosealicense.com/licenses/mit/)

pypdsadmin/__init__.py

@@ -0,0 +1 @@
+"""PDS Admin CLI."""

pypdsadmin/commands/__init__.py

@@ -0,0 +1,6 @@
+"""Module for CLI commands."""
+
+from .create_invite_code import CreateInviteCodeCommand
+from .request_crawl import RequestCrawlCommand
+
+__all__ = ["CreateInviteCodeCommand", "RequestCrawlCommand"]

pypdsadmin/commands/account/__init__.py

@@ -0,0 +1,19 @@
+"""Account Commands."""
+
+from .create import AccountCreateCommand
+from .delete import AccountDeleteCommand
+from .list import AccountListCommand
+from .reset_password import AccountResetPasswordCommand
+from .takedown import AccountTakedownCommand
+from .untakedown import AccountUntakedownCommand
+from .update_handle import AccountUpdateHandleCommand
+
+__all__ = [
+    "AccountCreateCommand",
+    "AccountDeleteCommand",
+    "AccountListCommand",
+    "AccountResetPasswordCommand",
+    "AccountTakedownCommand",
+    "AccountUntakedownCommand",
+    "AccountUpdateHandleCommand",
+]

pypdsadmin/commands/account/create.py

@@ -0,0 +1,98 @@
+"""Command to create a new account in the PDS server."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import requests
+from cleo.commands.command import Command
+from cleo.helpers import argument, option
+
+from pypdsadmin.consts import DEFAULT_PASSWORD_LENGTH, DEFAULT_TIMEOUT
+from pypdsadmin.env import environ
+from pypdsadmin.lib.tokens import generate_secure_token
+
+if TYPE_CHECKING:
+    from cleo.io.inputs.argument import Argument
+    from cleo.io.inputs.option import Option
+
+
+class AccountCreateCommand(Command):
+    """Command to create a new account in the PDS server."""
+
+    name = "account create"
+    description = "Create a new account in the PDS server."
+
+    arguments: list[Argument] = [  # noqa: RUF012
+        argument("email", description="Email address for the new account."),
+        argument("handle", description="Handle for the new account."),
+    ]
+
+    options: list[Option] = [  # noqa: RUF012
+        option(
+            "timeout",
+            "t",
+            description="The request timeout in seconds.",
+            flag=False,
+            default=DEFAULT_TIMEOUT,
+        ),
+        option(
+            "prompt-password",
+            "-P",
+            description="Prompt for a password instead of using a generated one.",
+            flag=True,
+        ),
+        option(
+            "password",
+            "-p",
+            description="Password for the new account.",
+            flag=False,
+            default=None,
+        ),
+    ]
+
+    def handle(self) -> None:
+        """Handle the command."""
+        email = self.argument("email")
+        handle = self.argument("handle")
+        hostname = environ["PDS_HOSTNAME"]
+        admin_password = environ["PDS_ADMIN_PASSWORD"]
+        timeout = int(self.option("timeout"))
+
+        if self.option("prompt-password"):
+            password = self.secret("Enter a password for the new account")
+        else:
+            password = self.option("password")
+            if not password:
+                password = generate_secure_token(DEFAULT_PASSWORD_LENGTH)
+
+        invite_response = requests.post(
+            f"https://{hostname}/xrpc/com.atproto.server.createInviteCode",
+            auth=("admin", admin_password),
+            json={"useCount": 1},
+            timeout=timeout,
+        )
+        invite_response.raise_for_status()
+        invite_code = invite_response.json()["code"]
+
+        account_response = requests.post(
+            f"https://{hostname}/xrpc/com.atproto.server.createAccount",
+            json={
+                "email": email,
+                "handle": handle,
+                "password": password,
+                "inviteCode": invite_code,
+            },
+            timeout=timeout,
+        )
+
+        try:
+            account_response.raise_for_status()
+        except requests.HTTPError:
+            json = account_response.json()
+            self.line(f"<error>Error: {json.get('error')} - {json.get('message')}</error>")
+            raise
+
+        did = account_response.json().get("did")
+
+        self.line(f"Account created successfully!\nHandle: {handle}\nDID: {did}\nPassword: {password}")

pypdsadmin/commands/account/delete.py

@@ -0,0 +1,69 @@
+"""Command to delete an account from the PDS server."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import requests
+from cleo.commands.command import Command
+from cleo.helpers import argument, option
+
+from pypdsadmin.consts import DEFAULT_TIMEOUT
+from pypdsadmin.env import environ
+
+if TYPE_CHECKING:
+    from cleo.io.inputs.argument import Argument
+    from cleo.io.inputs.option import Option
+
+
+class AccountDeleteCommand(Command):
+    """Command to delete an account from the PDS server."""
+
+    name = "account delete"
+    description = "Delete an account from the PDS server."
+
+    arguments: list[Argument] = [  # noqa: RUF012
+        argument("did", description="DID of the account to delete."),
+    ]
+
+    options: list[Option] = [  # noqa: RUF012
+        option(
+            "timeout",
+            "t",
+            description="The request timeout in seconds.",
+            flag=False,
+            default=DEFAULT_TIMEOUT,
+        ),
+    ]
+
+    def handle(self) -> None:
+        """Handle the command."""
+        did = self.argument("did")
+        if not did.startswith("did:"):
+            self.line_error("<error>DID must start with 'did:'</error>")
+            return
+
+        hostname = environ["PDS_HOSTNAME"]
+        admin_password = environ["PDS_ADMIN_PASSWORD"]
+        timeout = int(self.option("timeout"))
+
+        confirmation = self.confirm(f"This action is permanent. Delete account {did}?", default=False)
+        if not confirmation:
+            self.line("<comment>Operation cancelled.</comment>")
+            return
+
+        response = requests.post(
+            f"https://{hostname}/xrpc/com.atproto.admin.deleteAccount",
+            auth=("admin", admin_password),
+            json={"did": did},
+            timeout=timeout,
+        )
+
+        try:
+            response.raise_for_status()
+        except requests.HTTPError:
+            json = response.json()
+            self.line(f"<error>Error: {json.get('error')} - {json.get('message')}</error>")
+            raise
+
+        self.line(f"<info>{did} deleted successfully.</info>")

pypdsadmin/commands/account/list.py

@@ -0,0 +1,70 @@
+"""Command to list accounts in the PDS server."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import requests
+from cleo.commands.command import Command
+from cleo.helpers import option
+
+from pypdsadmin.consts import DEFAULT_TIMEOUT
+from pypdsadmin.env import environ
+
+if TYPE_CHECKING:
+    from cleo.io.inputs.option import Option
+
+
+class AccountListCommand(Command):
+    """Command to list accounts in the PDS server."""
+
+    name = "account list"
+    description = "List accounts in the PDS server."
+
+    options: list[Option] = [  # noqa: RUF012
+        option("timeout", "t", description="The request timeout in seconds.", flag=False, default=DEFAULT_TIMEOUT),
+        option("compact", "c", description="Compact the output.", flag=True),
+    ]
+
+    def handle(self) -> None:
+        """Handle the command."""
+        hostname = environ["PDS_HOSTNAME"]
+        self.write("<comment>Fetching repos...</comment>")
+        response = requests.get(
+            f"https://{hostname}/xrpc/com.atproto.sync.listRepos?limit=100",
+            timeout=int(self.option("timeout")),
+        )
+        try:
+            response.raise_for_status()
+        except requests.HTTPError:
+            json = response.json()
+            self.line(f"<error>Error: {json.get('error')} - {json.get('message')}</error>")
+            raise
+        repos = response.json().get("repos", [])
+
+        output = []
+        for repo in repos:
+            self.overwrite(f"<comment>Fetching account info for {repo.get('did')}...</comment>")
+            did = repo.get("did")
+            account_response = requests.get(
+                f"https://{hostname}/xrpc/com.atproto.admin.getAccountInfo?did={did}",
+                auth=("admin", environ["PDS_ADMIN_PASSWORD"]),
+                timeout=int(self.option("timeout")),
+            )
+            try:
+                account_response.raise_for_status()
+            except requests.HTTPError:
+                json = account_response.json()
+                self.line(f"<error>Error: {json.get('error')} - {json.get('message')}</error>")
+                raise
+            output.append(account_response.json())
+
+        if len(output) == 0:
+            self.overwrite("<info>No accounts found.</info>")
+            return
+        self.overwrite("")
+        self.render_table(
+            ["Handle", "Email", "DID"],  # type: ignore  # noqa: PGH003
+            [[row.get("handle", ""), row.get("email", ""), row.get("did", "")] for row in output],
+            style="compact" if self.option("compact") else "default",
+        )