SocialPub/tools/pasture/town/dialects/gts.py

"""GoToSocial 0.22: accounts made with its admin CLI (then one restart, as the pasture does), tokens through the OAuth
code flow with a signed-in cookie (it has no password grant), objects read from a copy of its sqlite database."""
import json
import re
import urllib.parse

from core import podman
from dialects.base import Stored
from dialects.mastodon_api import MastodonApi

# gtsmodel.Visibility since 0.20: stored as small integers
VIS = {2: "public", 3: "unlisted", 4: "followers", 5: "mutuals", 6: "direct"}
OOB = "urn:ietf:wg:oauth:2.0:oob"


class Gts(MastodonApi):
	platform = "gts"

	def provision(self, accounts):
		created = False
		existing = {r["username"] for r in self._sqlite("select username from accounts where domain is null")}
		for a in accounts:
			if a.username in existing:
				continue
			podman.exec_("pasture-gts", "/gotosocial/gotosocial", "admin", "account", "create", "--username", a.username,
						 "--email", f"{a.username}@gts.test", "--password", a.password)
			podman.exec_("pasture-gts", "/gotosocial/gotosocial", "admin", "account", "confirm", "--username", a.username)
			created = True
		if created:
			# the running server keeps what it has cached about accounts; the pasture restarts it after the CLI too
			podman.run("restart", "pasture-gts")
			self._wait()
		return [self.session_from_token(a, self._token(a)) for a in accounts]

	def _wait(self):
		import time
		for _ in range(90):
			try:
				if self.http.get(self.base + "/api/v1/instance").status == 200:
					return
			except Exception:
				pass
			time.sleep(1)
		raise TimeoutError("GoToSocial did not come back")

	def _token(self, a):
		app = self.http.post(self.base + "/api/v1/apps", ok={200}, form={
			"client_name": "pasture-town", "redirect_uris": OOB, "scopes": "read write follow"}).json()
		cookies = {}

		def send(method, path, **kw):
			# the sign-in pages are HTML: asked for JSON (the client's default) they answer 406 and set no session
			headers = {"Accept": "text/html,*/*"}
			if cookies:
				headers["Cookie"] = "; ".join(f"{k}={v}" for k, v in cookies.items())
			r = self.http.request(method, self.base + path, headers=headers, **kw)
			for k, v in r.headers:
				if k.lower() == "set-cookie":
					name, _, value = v.split(";")[0].partition("=")
					cookies[name.strip()] = value.strip()
			return r

		query = urllib.parse.urlencode({"client_id": app["client_id"], "redirect_uri": OOB, "response_type": "code",
										"scope": "read write follow"})
		send("GET", f"/oauth/authorize?{query}")
		send("POST", "/auth/sign_in", form={"username": f"{a.username}@gts.test", "password": a.password})
		r = send("POST", "/oauth/authorize")
		location = r.header("Location") or ""
		code = urllib.parse.parse_qs(urllib.parse.urlsplit(location).query).get("code", [None])[0]
		if code is None:
			found = re.search(r"code=([A-Za-z0-9_\-]+)", r.text)
			code = found.group(1) if found else None
		if code is None:
			raise RuntimeError(f"GoToSocial gave {a.username} no authorization code ({r.status})")
		token = self.http.post(self.base + "/oauth/token", ok={200}, form={
			"grant_type": "authorization_code", "code": code, "client_id": app["client_id"],
			"client_secret": app["client_secret"], "redirect_uri": OOB, "scope": "read write follow"}).json()
		return token["access_token"]

	def _sqlite(self, sql, *params):
		with podman.SqliteCopy("pasture-gts", "/gotosocial/storage/sqlite.db") as db:
			return db.rows(sql, *params)

	def _rows(self, uris):
		if not uris:
			return {}
		marks = ",".join("?" * len(uris))
		rows = self._sqlite(f"""
			select s.id as local_id, s.uri, s.visibility, s.flags, s.text, s.content, s.content_warning as cw,
				s.edited_at is not null as edited, s.in_reply_to_uri as parent_uri,
				(select count(*) from status_faves f where f.status_id = s.id) as likes,
				(select count(*) from statuses b where b.boost_of_id = s.id) as boosts,
				(select count(*) from statuses r where r.in_reply_to_id = s.id) as replies,
				(select votes from polls p where p.status_id = s.id) as votes
			from statuses s where s.boost_of_id is null and s.uri in ({marks})""", *uris)
		out = {}
		for r in rows:
			votes = json.loads(r["votes"]) if r["votes"] else None
			# 0.22 soft-deletes: flag 2 marks a deleted status (its statuses_deleted_idx), 8 a local one
			out[r["uri"]] = Stored(True, bool(r["flags"] & 2), r["local_id"], VIS.get(r["visibility"], str(r["visibility"])),
								   r["text"] or r["content"], r["cw"] or None, bool(r["edited"]), r["parent_uri"],
								   r["likes"], r["boosts"], r["replies"], votes, None, r)
		return out