Michał Szczepanik
22498a7408
This test is longer and has more of a workflow (set-up, push, clone, get) than the other tests which check more isolated / well defined behaviors, so it seems appropriate to put it in a separate file called test_workflows. This test is also unique in that it relies on gpg and requires a gpg key to be specified (by id), and gets skipped if the `--gpg-keyid` argument is not provided (or if gpg does not have a matching private key). The idea is to have a throwaway key ready and gpg set up. I was looking for maybe using stateless (thinking we might add the throwaway key to the tests repo), but git-annex only supports it for the shared mode (unencrypted cipher in git repo), which would not make sense with forgejo. The assertions are perhaps a little redundant (metadata-based availability from whereis, remote checks with checkpresentkey, trying to get files) but I wasn't sure what would be the best set of checks.
184 lines
7.4 KiB
Python
184 lines
7.4 KiB
Python
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
|
|
from datalad_core.clone import clone_annexrepo
|
|
from datalad_core.create import create_annexrepo
|
|
from datalad_core.runners import call_annex_json_lines, call_git
|
|
from datalad_core.runners import CommandError
|
|
import pytest
|
|
|
|
|
|
def test_mask(pytestconfig: pytest.Config, monkeypatch, forgejo, words):
|
|
"""Test a mask special remote with encryption
|
|
|
|
The mask special remote allows setting up encryption for (some)
|
|
contents sent to Forgejo.
|
|
|
|
Here we create a set-up with one file sent to Git, one to regular
|
|
annnex, and one to masked / encrypted annex in forgejo (using
|
|
preferred content). After pushing, we clone, and perform checks
|
|
in the clone. We make assertions for availability information
|
|
(whereis), test key presence (checkpresentkey), and finally run
|
|
get (first mocking the absence of GPG key and then normally).
|
|
|
|
This test assumes having gpg available, and uses a key ID provided
|
|
as an argument to pytest. It gets skipped if either of those is
|
|
not met. It is best to have a throwaway key with no password, to
|
|
avoid interaction.
|
|
|
|
Note: git-annex supports stateless OpenPGP, which could be nice
|
|
for test setup, but only with encryption=shared, which is
|
|
pointless for a git+annex setup with Forgejo.
|
|
|
|
See:
|
|
- https://git-annex.branchable.com/special_remotes/mask/
|
|
- https://rdm.sfb1451.de/walkthroughs/forgejo-encryption/index.html
|
|
- https://git-annex.branchable.com/todo/support_using_Stateless_OpenPGP_instead_of_gpg/
|
|
|
|
"""
|
|
# check if key ID was provided and gpg can access it
|
|
keyid = pytestconfig.getoption("gpg_keyid", skip=True) # skip if no key id
|
|
try:
|
|
_ = subprocess.run(args=["gpg", "--list-secret-keys", keyid], check=True) # type: ignore
|
|
except subprocess.CalledProcessError:
|
|
pytest.skip("GPG key with provided ID not present")
|
|
|
|
# create a Forgejo repo
|
|
repo_name = words.random_name()
|
|
org: str | None = pytestconfig.getoption("org")
|
|
r = forgejo.create_repo(name=repo_name, private=False, org=org)
|
|
clone_url = r.json()["clone_url"]
|
|
|
|
# create a local repository, set up remote+mask, and push
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = create_annexrepo(repo_path, private=True)
|
|
|
|
# create three files, with different purposes later on
|
|
git_file = wt.path / "README"
|
|
annex_ordinary_file = wt.path / "foo.txt"
|
|
annex_protected_file = wt.path / "protected_bar.txt"
|
|
|
|
git_file.write_text("Lorem ipsum")
|
|
annex_ordinary_file.write_text("Lorem ipsum dolor sit amet")
|
|
annex_protected_file.write_text("Ut enim ad minim veniam")
|
|
|
|
# add one file to git
|
|
_ = call_git(args=["add", git_file.name], cwd=wt.path)
|
|
# add two files to annex and keep track of the annex keys
|
|
file_keys = dict()
|
|
for res in call_annex_json_lines(
|
|
annex_args=["add", annex_ordinary_file.name, annex_protected_file.name],
|
|
cwd=wt.path,
|
|
):
|
|
file_keys[res["file"]] = res["key"]
|
|
|
|
# commit, add remote, and push (git only) so remote has uuid but no annex keys yet
|
|
_ = call_git(args=["commit", "-m", "Add files"], cwd=wt.path)
|
|
_ = call_git(args=["remote", "add", "origin", clone_url], cwd=wt.path)
|
|
_ = call_git(args=["annex", "push", "origin", "--no-content"], cwd=wt.path)
|
|
|
|
# initalize the mask special remote, encryption=hybrid (default)
|
|
_ = call_git(
|
|
args=[
|
|
"annex",
|
|
"initremote",
|
|
"encrypted-origin",
|
|
"type=mask",
|
|
"remote=origin",
|
|
"encryption=hybrid",
|
|
f"keyid={keyid}",
|
|
],
|
|
cwd=wt.path,
|
|
)
|
|
# set wanted (one file to origin, one to mask) and push both remotes
|
|
_ = call_git(
|
|
args=["annex", "wanted", "origin", "exclude=protected*"], cwd=wt.path
|
|
)
|
|
_ = call_git(
|
|
args=["annex", "wanted", "encrypted-origin", "include=protected*"],
|
|
cwd=wt.path,
|
|
)
|
|
_ = call_git(args=["annex", "push"], cwd=wt.path)
|
|
|
|
# perform checks in a clone
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir)
|
|
wt = clone_annexrepo(clone_url, repo_path)
|
|
_ = call_git(args=["annex", "enableremote", "encrypted-origin"], cwd=wt.path)
|
|
|
|
# get UUIDs
|
|
uuid_origin = next(
|
|
call_annex_json_lines(annex_args=["info", "origin"], cwd=wt.path)
|
|
)["uuid"]
|
|
uuid_mask = next(
|
|
call_annex_json_lines(annex_args=["info", "encrypted-origin"], cwd=wt.path)
|
|
)["uuid"]
|
|
|
|
# whereis: report on logged local information
|
|
# check whether files went to correct remotes
|
|
for res in call_annex_json_lines(annex_args=["whereis"], cwd=wt.path):
|
|
loc_uuids = {x["uuid"] for x in res["whereis"]}
|
|
if res["file"] == annex_ordinary_file.name:
|
|
# no subdirs, so old .name is fine
|
|
assert uuid_origin in loc_uuids
|
|
elif res["file"] == annex_protected_file.name:
|
|
assert uuid_mask in loc_uuids and uuid_origin not in loc_uuids
|
|
|
|
# checkpresentkey: actively check the remote
|
|
# checkpresentkey exits with status 0 (present) or 1 (absent)
|
|
# call_git raises on non-zero status
|
|
# the ordinary file should be in the unencrypted remote
|
|
_ = call_git(
|
|
args=[
|
|
"annex",
|
|
"checkpresentkey",
|
|
file_keys[annex_ordinary_file.name],
|
|
"origin",
|
|
],
|
|
cwd=wt.path,
|
|
)
|
|
# the protected file should be in the encrypted remote
|
|
_ = call_git(
|
|
args=[
|
|
"annex",
|
|
"checkpresentkey",
|
|
file_keys[annex_protected_file.name],
|
|
"encrypted-origin",
|
|
],
|
|
cwd=wt.path,
|
|
)
|
|
# the protected file should not be in the unencrypted remote
|
|
with pytest.raises(CommandError):
|
|
_ = call_git(
|
|
args=[
|
|
"annex",
|
|
"checkpresentkey",
|
|
file_keys[annex_protected_file.name],
|
|
"origin",
|
|
],
|
|
cwd=wt.path,
|
|
)
|
|
|
|
# high level test: can't get encrypted without GPG key
|
|
with tempfile.TemporaryDirectory() as fakehome:
|
|
with monkeypatch.context() as m:
|
|
m.setenv("GNUPGHOME", fakehome)
|
|
with pytest.raises(CommandError):
|
|
# call_annex_json_lines yields results & raises at the end
|
|
for res in call_annex_json_lines(annex_args=["get"], cwd=wt.path):
|
|
if res["file"] == annex_ordinary_file.name:
|
|
assert res["success"]
|
|
elif res["file"] == annex_protected_file.name:
|
|
assert not res["success"]
|
|
# drop the files
|
|
for res in call_annex_json_lines(annex_args=["drop"], cwd=wt.path):
|
|
pass
|
|
# and with the key available again, we should be able to get all
|
|
for res in call_annex_json_lines(annex_args=["get"], cwd=wt.path):
|
|
assert res["success"]
|
|
|
|
# and just to be sure, check if we can read plaintext
|
|
content = (wt.path / annex_protected_file.name).read_text()
|
|
assert content.rstrip() == "Ut enim ad minim veniam"
|