103 lines
4.5 KiB
Python
103 lines
4.5 KiB
Python
 
|
"""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
|