297 lines
8.5 KiB
Python
297 lines
8.5 KiB
Python
import io
|
|
import json
|
|
import typing as t
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import inquirer
|
|
import typer
|
|
from rich.console import Console
|
|
|
|
try:
|
|
from importlib.metadata import version
|
|
except ImportError:
|
|
from importlib_metadata import version
|
|
|
|
__version__ = version("oxo")
|
|
|
|
app = typer.Typer()
|
|
|
|
BASE_URL = typer.Option("https://0x0.st", envvar="OXO_BASE_URL")
|
|
TOKEN_CACHE_DIR: t.Optional[str] = typer.Option(None, envvar="OXO_TOKEN_CACHE_DIR")
|
|
|
|
err_console = Console(stderr=True, color_system=None)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TokenData:
|
|
token: str
|
|
oxo_url: str
|
|
|
|
@property
|
|
def curl_rm(self):
|
|
return f"curl -s -Ftoken={self.token} -Fdelete= {self.oxo_url}"
|
|
|
|
@property
|
|
def curl_expiry(self, expiry_time="NUMHOURS"):
|
|
return f"curl -s -Ftoken={self.token} -Fexpires={expiry_time} {self.oxo_url}"
|
|
|
|
|
|
def post_to(
|
|
base_url: str,
|
|
*,
|
|
data: t.Iterable[t.Dict],
|
|
expires: t.Optional[int],
|
|
token_dir: t.Optional[Path],
|
|
use_long_name: bool,
|
|
) -> str:
|
|
retval = []
|
|
with httpx.Client() as client:
|
|
data_dict = {}
|
|
if expires:
|
|
data_dict["expires"] = expires
|
|
if use_long_name:
|
|
data_dict["secret"] = use_long_name
|
|
for d in data:
|
|
if isinstance(d.get("file"), io.BufferedReader):
|
|
res = client.post(base_url, files=d, data=data_dict)
|
|
else:
|
|
res = client.post(base_url, data={**d, **data_dict})
|
|
res.raise_for_status()
|
|
remote_url = res.text.strip()
|
|
token = res.headers.get("x-token")
|
|
if token:
|
|
token_data = TokenData(
|
|
token=token,
|
|
oxo_url=remote_url,
|
|
)
|
|
err_console.print(f"To remove post, {token_data.curl_rm}")
|
|
err_console.print(
|
|
f"To update expiration date, {token_data.curl_expiry}"
|
|
)
|
|
if token_dir:
|
|
# todo rewrite when drop 3.7
|
|
if d.get("url"):
|
|
fname = d.get("url").rpartition("/")[2]
|
|
elif d.get("file").name:
|
|
fname = Path(d.get("file").name).name
|
|
else:
|
|
raise ValueError(f"{d}")
|
|
token_dir.joinpath(f"{fname}.token").write_text(
|
|
json.dumps(asdict(token_data))
|
|
)
|
|
|
|
retval.append(remote_url)
|
|
return " ".join(retval)
|
|
|
|
|
|
def post_files(
|
|
base_url,
|
|
files: t.List[Path],
|
|
*,
|
|
expires: t.Optional[int],
|
|
token_dir: t.Optional[Path],
|
|
use_long_name: bool,
|
|
):
|
|
return post_to(
|
|
base_url,
|
|
data=({"file": f.open("rb")} for f in files),
|
|
expires=expires,
|
|
token_dir=token_dir,
|
|
use_long_name=use_long_name,
|
|
)
|
|
|
|
|
|
def post_repost(
|
|
base_url,
|
|
urls: t.List[str],
|
|
*,
|
|
expires: t.Optional[int],
|
|
token_dir: t.Optional[Path],
|
|
use_long_name: bool,
|
|
):
|
|
return post_to(
|
|
base_url,
|
|
data=({"url": u.strip()} for u in urls),
|
|
expires=expires,
|
|
token_dir=token_dir,
|
|
use_long_name=use_long_name,
|
|
)
|
|
|
|
|
|
def post_shorten(
|
|
base_url,
|
|
urls: t.List[str],
|
|
*,
|
|
expires: t.Optional[int],
|
|
token_dir: t.Optional[Path],
|
|
):
|
|
return post_to(
|
|
base_url,
|
|
data=({"shorten": u.strip()} for u in urls),
|
|
expires=expires,
|
|
token_dir=token_dir,
|
|
use_long_name=False,
|
|
)
|
|
|
|
|
|
def version_callback(value: bool):
|
|
if value:
|
|
typer.echo(f"{__version__}")
|
|
raise typer.Exit()
|
|
|
|
|
|
@app.callback()
|
|
def main(
|
|
version: t.Optional[bool] = typer.Option(
|
|
None,
|
|
"--version",
|
|
callback=version_callback,
|
|
is_eager=True,
|
|
help="Show the version and exit.",
|
|
),
|
|
):
|
|
"""A command line utility for 0x0.st compliant pastebins.
|
|
|
|
To use a different 0x0 site, set `OXO_BASE_URL` in your environment,
|
|
or specify it in the relevant subcommand.
|
|
"""
|
|
|
|
|
|
def configure_token_dir(token_cache_dir: Path) -> Path:
|
|
if token_cache_dir:
|
|
token_dir = Path(token_cache_dir).expanduser().resolve()
|
|
if not token_dir.is_dir():
|
|
typer.secho(
|
|
f"token_dir={token_dir} is not a directory; cannot store tokens here.",
|
|
fg=typer.colors.RED,
|
|
err=True,
|
|
)
|
|
raise typer.Exit(-1)
|
|
token_dir.mkdir(exist_ok=True, parents=True)
|
|
else:
|
|
token_dir = None
|
|
return token_dir
|
|
|
|
|
|
@app.command()
|
|
def files(
|
|
files: t.List[Path] = typer.Argument(
|
|
..., min=1, exists=True, file_okay=True, dir_okay=False, resolve_path=True
|
|
),
|
|
use_long_name: bool = typer.Option(
|
|
False, help="If true, create a harder to guess name."
|
|
),
|
|
base_url: str = BASE_URL,
|
|
expires: t.Optional[int] = typer.Option(
|
|
None, help="Expiration time, in hours or epoch milliseconds"
|
|
),
|
|
token_cache_dir: str = TOKEN_CACHE_DIR,
|
|
):
|
|
"""Upload one or more files. If one is provided by the 0x0 site used on a
|
|
successful upload, a management token will be printed to stderr along with
|
|
instructions on how to use it to delete or adjust the expiration time on
|
|
the uploaded file(s). If `--token-cache-dir` is passed or
|
|
`OXO_TOKEN_CACHE_DIR` is set, the management token will also be cached as a
|
|
json file in that directory. The cache directory will be created on first
|
|
use.
|
|
|
|
"""
|
|
token_dir = configure_token_dir(token_cache_dir)
|
|
typer.echo(
|
|
post_files(
|
|
base_url,
|
|
files,
|
|
expires=expires,
|
|
token_dir=token_dir,
|
|
use_long_name=use_long_name,
|
|
)
|
|
)
|
|
|
|
|
|
@app.command()
|
|
def repost(
|
|
urls: t.List[str] = typer.Argument(..., min=1),
|
|
use_long_name: bool = typer.Option(
|
|
False, help="If true, create a harder to guess name."
|
|
),
|
|
base_url=BASE_URL,
|
|
expires: t.Optional[int] = typer.Option(
|
|
None, help="Expiration time, in hours or epoch milliseconds"
|
|
),
|
|
token_cache_dir: str = TOKEN_CACHE_DIR,
|
|
):
|
|
"""Repost one or more urls."""
|
|
token_dir = configure_token_dir(token_cache_dir)
|
|
typer.echo(
|
|
post_repost(
|
|
base_url,
|
|
urls,
|
|
expires=expires,
|
|
token_dir=token_dir,
|
|
use_long_name=use_long_name,
|
|
)
|
|
)
|
|
|
|
|
|
@app.command()
|
|
def shorten(
|
|
urls: t.List[str] = typer.Argument(..., min=1),
|
|
base_url=BASE_URL,
|
|
expires: t.Optional[int] = typer.Option(
|
|
None, help="Expiration time, in hours or epoch milliseconds"
|
|
),
|
|
token_cache_dir: str = TOKEN_CACHE_DIR,
|
|
):
|
|
"""Shorten one or more urls."""
|
|
token_dir = configure_token_dir(token_cache_dir)
|
|
if base_url == BASE_URL.default:
|
|
typer.secho(
|
|
f"Warning: shortening is often disabled "
|
|
f"for {base_url}, command may fail",
|
|
fg=typer.colors.RED,
|
|
err=True,
|
|
)
|
|
typer.echo(post_shorten(base_url, urls, expires, token_dir))
|
|
|
|
|
|
def did_confirm(token: TokenData, filename: Path) -> bool:
|
|
confirm = inquirer.Confirm(
|
|
"delete", message=f"delete {filename.name} ({token.oxo_url})?", default=True
|
|
)
|
|
res = inquirer.prompt([confirm])
|
|
return res["delete"]
|
|
|
|
|
|
@app.command()
|
|
def delete(
|
|
token_cache_dir: Path = typer.Argument("", envvar="OXO_TOKEN_CACHE_DIR"),
|
|
interactive: bool = typer.Option(
|
|
False, help="If True, prompt for each token found."
|
|
),
|
|
):
|
|
"""Delete uploaded files if tokens are found on disk at `token_cache_dir`."""
|
|
token_files = token_cache_dir.glob("*.token")
|
|
tokens = []
|
|
good_tokens = []
|
|
for token in token_files:
|
|
try:
|
|
tokens.append(TokenData(**json.loads(token.read_text())))
|
|
good_tokens.append(token)
|
|
except Exception as e:
|
|
typer.secho(f"Ignoring {token}: {e}", err=True, fg=typer.colors.YELLOW)
|
|
with httpx.Client() as client:
|
|
for token, token_file in zip(tokens, good_tokens):
|
|
try:
|
|
if (interactive and did_confirm(token, token_file)) or not interactive:
|
|
res = client.post(
|
|
token.oxo_url, data=dict(token=token.token, delete="")
|
|
)
|
|
res.raise_for_status()
|
|
typer.echo(f"Removed {token.oxo_url} {res.text.strip()}")
|
|
token_file.unlink()
|
|
typer.echo(f"Removed stale token {token_file}")
|
|
except Exception as e:
|
|
typer.secho(e, err=True, fg=typer.colors.RED)
|