91 lines
4.5 KiB
Python
91 lines
4.5 KiB
Python
 
|
"""Iceshrimp.NET 2026.1: accounts through its own API (/api/iceshrimp/auth/register, registrations open in the pasture),
|
||
|
|
Mastodon API tokens through its OAuth page, which takes the user's name and password itself (an ASP.NET form with an
|
||
|
|
antiforgery token). Its database keeps Misskey's shape (`note`, `poll`), with likes counted apart from reactions."""
|
||
|
|
import html
|
||
|
|
import re
|
||
|
|
import urllib.parse
|
||
|
|
|
||
|
|
from core import podman
|
||
|
|
from dialects.base import Stored
|
||
|
|
from dialects.mastodon_api import MastodonApi
|
||
|
|
|
||
|
|
OOB = "urn:ietf:wg:oauth:2.0:oob"
|
||
|
|
VIS = {"public": "public", "home": "unlisted", "followers": "followers", "specified": "direct"}
|
||
|
|
|
||
|
|
|
||
|
|
class Iceshrimp(MastodonApi):
|
||
|
|
platform = "iceshrimp"
|
||
|
|
caps = MastodonApi.caps | {"quote", "react"}
|
||
|
|
|
||
|
|
def provision(self, accounts):
|
||
|
|
app = self.http.post(self.base + "/api/v1/apps", ok={200}, form={
|
||
|
|
"client_name": "pasture-town", "redirect_uris": OOB, "scopes": "read write follow"}).json()
|
||
|
|
sessions = []
|
||
|
|
existing = {r["username"] for r in podman.psql("iceshrimp", 'select username from "user" where host is null')}
|
||
|
|
for a in accounts:
|
||
|
|
# its login answers an unknown name by cutting the connection short, so only new names are registered
|
||
|
|
if a.username not in existing:
|
||
|
|
self.http.post(self.base + "/api/iceshrimp/auth/register", ok={200},
|
||
|
|
json={"username": a.username, "password": a.password})
|
||
|
|
sessions.append(self.session_from_token(a, self._token(app, a)))
|
||
|
|
return sessions
|
||
|
|
|
||
|
|
def _token(self, app, a):
|
||
|
|
cookies = {}
|
||
|
|
|
||
|
|
def send(method, path, form=None):
|
||
|
|
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, form=form)
|
||
|
|
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"})
|
||
|
|
page = send("GET", f"/oauth/authorize?{query}").text
|
||
|
|
antiforgery = re.search(r'name="__RequestVerificationToken" value="([^"]+)"', page).group(1)
|
||
|
|
r = send("POST", f"/oauth/authorize?{query}", {
|
||
|
|
"_handler": "oauth-authorize", "__RequestVerificationToken": html.unescape(antiforgery),
|
||
|
|
"Model.Username": a.username, "Model.Password": a.password})
|
||
|
|
location = r.header("Location") or ""
|
||
|
|
code = urllib.parse.parse_qs(urllib.parse.urlsplit(location).query).get("code", [None])[0]
|
||
|
|
if code is None:
|
||
|
|
# the out-of-band page says "Your code is: <code>" in its text
|
||
|
|
text = re.sub(r"<[^>]+>", " ", re.sub(r"<(style|script).*?</\1>", "", r.text, flags=re.S))
|
||
|
|
found = re.search(r"Your code is:\s*([A-Za-z0-9_\-]{16,})", text)
|
||
|
|
code = found.group(1) if found else None
|
||
|
|
if code is None:
|
||
|
|
raise RuntimeError(f"Iceshrimp gave {a.username} no authorization code ({r.status}): {r.text[:300]}")
|
||
|
|
return 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()["access_token"]
|
||
|
|
|
||
|
|
def react(self, s, uri, emoji):
|
||
|
|
sid = self.local_status_id(s, uri) or self.resolve(s, uri)
|
||
|
|
self.api(s, "POST", f"/api/v1/statuses/{sid}/react/{urllib.parse.quote(emoji)}", ok={200})
|
||
|
|
|
||
|
|
def _rows(self, uris):
|
||
|
|
if not uris:
|
||
|
|
return {}
|
||
|
|
local = {u: u.rsplit("/", 1)[1] for u in uris if u.startswith(f"{self.base}/notes/")}
|
||
|
|
rows = podman.psql("iceshrimp", f"""
|
||
|
|
select n.id, n.uri, n.visibility::text as visibility, n.text, n.cw, n."updatedAt" is not null as edited,
|
||
|
|
n."renoteCount" as boosts, n."repliesCount" as replies, n."likeCount" as likes, n.reactions, p.votes,
|
||
|
|
coalesce(r.uri, case when r.id is not null then '{self.base}/notes/' || r.id end) as parent_uri
|
||
|
|
from note n left join poll p on p."noteId" = n.id left join note r on r.id = n."replyId"
|
||
|
|
where (n."renoteId" is null or n.text is not null)
|
||
|
|
and (n.uri = any(string_to_array(:'p1', ' ')) or n.id = any(string_to_array(:'p2', ' ')))""",
|
||
|
|
" ".join(uris), " ".join(local.values()) or "-")
|
||
|
|
out = {}
|
||
|
|
for r in rows:
|
||
|
|
uri = r["uri"] or f"{self.base}/notes/{r['id']}"
|
||
|
|
if uri not in uris:
|
||
|
|
continue
|
||
|
|
out[uri] = Stored(True, False, r["id"], VIS.get(r["visibility"], r["visibility"]), r["text"], r["cw"],
|
||
|
|
bool(r["edited"]), r["parent_uri"], r["likes"], r["boosts"], r["replies"], r["votes"],
|
||
|
|
r["reactions"] or {}, r)
|
||
|
|
return out
|