Adapt session tests to new dev env (nspcc-dev#714)

dynamic_env_pytest_tests/lib/helpers/session_token.py

@@ -0,0 +1,286 @@
+import base64
+import json
+import logging
+import os
+import uuid
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Optional
+
+import allure
+import json_transformers
+from common import ASSETS_DIR, NEOFS_CLI_EXEC, TEST_FILES_DIR, WALLET_CONFIG
+from data_formatters import get_wallet_public_key
+from json_transformers import encode_for_json
+from neofs_testlib.cli import NeofsCli
+from neofs_testlib.env.env import NodeWallet
+from neofs_testlib.shell import Shell
+
+from helpers.storage_object_info import StorageObjectInfo
+
+logger = logging.getLogger("NeoLogger")
+
+UNRELATED_KEY = "unrelated key in the session"
+UNRELATED_OBJECT = "unrelated object in the session"
+UNRELATED_CONTAINER = "unrelated container in the session"
+WRONG_VERB = "wrong verb of the session"
+INVALID_SIGNATURE = "invalid signature of the session data"
+
+
+class ObjectVerb(Enum):
+    PUT = "PUT"
+    DELETE = "DELETE"
+    GET = "GET"
+    RANGEHASH = "RANGEHASH"
+    RANGE = "RANGE"
+    HEAD = "HEAD"
+    SEARCH = "SEARCH"
+
+
+class ContainerVerb(Enum):
+    CREATE = "PUT"
+    DELETE = "DELETE"
+    SETEACL = "SETEACL"
+
+
+@dataclass
+class Lifetime:
+    exp: int = 100000000
+    nbf: int = 0
+    iat: int = 0
+
+
+@allure.step("Generate Session Token")
+def generate_session_token(
+    owner_wallet: NodeWallet,
+    session_wallet: NodeWallet,
+    session: dict[str, dict[str, Any]],
+    tokens_dir: str,
+    lifetime: Optional[Lifetime] = None,
+) -> str:
+    """
+    This function generates session token  and writes it to the file.
+    Args:
+        owner_wallet: wallet of container owner
+        session_wallet: wallet to which we grant the access via session token
+        session: Contains allowed operation with parameters
+        tokens_dir: Dir for token
+        lifetime: lifetime options for session
+    Returns:
+        The path to the generated session token file
+    """
+
+    file_path = os.path.join(tokens_dir, str(uuid.uuid4()))
+
+    pub_key_64 = get_wallet_public_key(session_wallet.path, session_wallet.password, "base64")
+
+    lifetime = lifetime or Lifetime()
+
+    session_token = {
+        "body": {
+            "id": f"{base64.b64encode(uuid.uuid4().bytes).decode('utf-8')}",
+            "ownerID": {
+                "value": f"{json_transformers.encode_for_json(owner_wallet.address)}"
+            },
+            "lifetime": {
+                "exp": f"{lifetime.exp}",
+                "nbf": f"{lifetime.nbf}",
+                "iat": f"{lifetime.iat}",
+            },
+            "sessionKey": pub_key_64,
+        }
+    }
+    session_token["body"].update(session)
+
+    logger.info(f"Got this Session Token: {session_token}")
+    with open(file_path, "w", encoding="utf-8") as session_token_file:
+        json.dump(session_token, session_token_file, ensure_ascii=False, indent=4)
+
+    return file_path
+
+
+@allure.step("Generate Session Token For Container")
+def generate_container_session_token(
+    owner_wallet: NodeWallet,
+    session_wallet: NodeWallet,
+    verb: ContainerVerb,
+    tokens_dir: str,
+    lifetime: Optional[Lifetime] = None,
+    cid: Optional[str] = None,
+) -> str:
+    """
+    This function generates session token for ContainerSessionContext
+    and writes it to the file. It is able to prepare session token file
+    for a specific container (<cid>) or for every container (adds
+    "wildcard" field).
+    Args:
+        owner_wallet: wallet of container owner.
+        session_wallet: wallet to which we grant the access via session token.
+        verb: verb to grant access to.
+        lifetime: lifetime options for session.
+        cid: container ID of the container
+    Returns:
+        The path to the generated session token file
+    """
+    session = {
+        "container": {
+            "verb": verb.value,
+            "wildcard": cid is None,
+            **({"containerID": {"value": f"{encode_for_json(cid)}"}} if cid is not None else {}),
+        },
+    }
+
+    return generate_session_token(
+        owner_wallet=owner_wallet,
+        session_wallet=session_wallet,
+        session=session,
+        tokens_dir=tokens_dir,
+        lifetime=lifetime,
+    )
+
+
+@allure.step("Generate Session Token For Object")
+def generate_object_session_token(
+    owner_wallet: NodeWallet,
+    session_wallet: NodeWallet,
+    oids: list[str],
+    cid: str,
+    verb: ObjectVerb,
+    tokens_dir: str,
+    lifetime: Optional[Lifetime] = None,
+) -> str:
+    """
+    This function generates session token for ObjectSessionContext
+    and writes it to the file.
+    Args:
+        owner_wallet: wallet of container owner
+        session_wallet: wallet to which we grant the access via session token
+        cid: container ID of the container
+        oids: list of objectIDs to put into session
+        verb: verb to grant access to; Valid verbs are: ObjectVerb.
+        lifetime: lifetime options for session
+    Returns:
+        The path to the generated session token file
+    """
+    session = {
+        "object": {
+            "verb": verb.value,
+            "target": {
+                "container": {"value": encode_for_json(cid)},
+                "objects": [{"value": encode_for_json(oid)} for oid in oids],
+            },
+        },
+    }
+
+    return generate_session_token(
+        owner_wallet=owner_wallet,
+        session_wallet=session_wallet,
+        session=session,
+        tokens_dir=tokens_dir,
+        lifetime=lifetime,
+    )
+
+
+@allure.step("Get signed token for container session")
+def get_container_signed_token(
+    owner_wallet: NodeWallet,
+    user_wallet: NodeWallet,
+    verb: ContainerVerb,
+    shell: Shell,
+    tokens_dir: str,
+    lifetime: Optional[Lifetime] = None,
+) -> str:
+    """
+    Returns signed token file path for static container session
+    """
+    session_token_file = generate_container_session_token(
+        owner_wallet=owner_wallet,
+        session_wallet=user_wallet,
+        verb=verb,
+        tokens_dir=tokens_dir,
+        lifetime=lifetime,
+    )
+    return sign_session_token(shell, session_token_file, owner_wallet)
+
+
+@allure.step("Get signed token for object session")
+def get_object_signed_token(
+    owner_wallet: NodeWallet,
+    user_wallet: NodeWallet,
+    cid: str,
+    storage_objects: list[StorageObjectInfo],
+    verb: ObjectVerb,
+    shell: Shell,
+    tokens_dir: str,
+    lifetime: Optional[Lifetime] = None,
+) -> str:
+    """
+    Returns signed token file path for static object session
+    """
+    storage_object_ids = [storage_object.oid for storage_object in storage_objects]
+    session_token_file = generate_object_session_token(
+        owner_wallet=owner_wallet,
+        session_wallet=user_wallet,
+        oids=storage_object_ids,
+        cid=cid,
+        verb=verb,
+        tokens_dir=tokens_dir,
+        lifetime=lifetime,
+    )
+    return sign_session_token(shell, session_token_file, owner_wallet)
+
+
+@allure.step("Create Session Token")
+def create_session_token(
+    shell: Shell,
+    owner: str,
+    wallet_path: str,
+    wallet_password: str,
+    rpc_endpoint: str,
+    lifetime: Optional[int] = None,
+    expire_at: Optional[int] = None,
+) -> str:
+    """
+    Create session token for an object.
+    Args:
+        shell: Shell instance.
+        owner: User that writes the token.
+        wallet_path: The path to wallet to which we grant the access via session token.
+        wallet_password: Wallet password.
+        rpc_endpoint: Remote node address (as 'multiaddr' or '<host>:<port>').
+    Returns:
+        The path to the generated session token file.
+    """
+    session_token = os.path.join(os.getcwd(), ASSETS_DIR, TEST_FILES_DIR, str(uuid.uuid4()))
+    neofscli = NeofsCli(shell=shell, neofs_cli_exec_path=NEOFS_CLI_EXEC)
+    neofscli.session.create(
+        rpc_endpoint=rpc_endpoint,
+        address=owner,
+        wallet=wallet_path,
+        wallet_password=wallet_password,
+        out=session_token,
+        lifetime=lifetime,
+        expire_at=expire_at,
+    )
+    return session_token
+
+
+@allure.step("Sign Session Token")
+def sign_session_token(shell: Shell, session_token_file: str, wlt: NodeWallet) -> str:
+    """
+    This function signs the session token by the given wallet.
+
+    Args:
+        shell: Shell instance.
+        session_token_file: The path to the session token file.
+        wlt: The path to the signing wallet.
+
+    Returns:
+        The path to the signed token.
+    """
+    signed_token_file = os.path.join(os.getcwd(), ASSETS_DIR, TEST_FILES_DIR, str(uuid.uuid4()))
+    neofscli = NeofsCli(shell=shell, neofs_cli_exec_path=NEOFS_CLI_EXEC, config_file=WALLET_CONFIG)
+    neofscli.util.sign_session_token(
+        wallet=wlt.path, from_file=session_token_file, to_file=signed_token_file
+    )
+    return signed_token_file

dynamic_env_pytest_tests/lib/helpers/storage_object_info.py

@@ -1,6 +1,24 @@
+import json
+import logging
 from dataclasses import dataclass
+from time import sleep
 from typing import Optional
 
+import allure
+import neofs_env.neofs_epoch as neofs_epoch
+import pytest
+from common import WALLET_PASS
+from grpc_responses import OBJECT_ALREADY_REMOVED
+from neo3.wallet import wallet
+from neofs_testlib.env.env import NeoFSEnv
+from neofs_testlib.shell import Shell
+from neofs_verbs import head_object
+from python_keywords.neofs_verbs import delete_object, get_object
+
+logger = logging.getLogger("NeoLogger")
+
+CLEANUP_TIMEOUT = 10
+
 
 @dataclass
 class ObjectRef:
@@ -23,3 +41,80 @@ class StorageObjectInfo(ObjectRef):
     attributes: Optional[list[dict[str, str]]] = None
     tombstone: Optional[str] = None
     locks: Optional[list[LockObjectInfo]] = None
+
+
+@allure.step("Verify Head Tombstone")
+def verify_head_tombstone(
+    wallet_path: str, cid: str, oid_ts: str, oid: str, shell: Shell, endpoint: str
+):
+    header = head_object(wallet_path, cid, oid_ts, shell=shell, endpoint=endpoint)["header"]
+
+    s_oid = header["sessionToken"]["body"]["object"]["target"]["objects"]
+    logger.info(f"Header Session OIDs is {s_oid}")
+    logger.info(f"OID is {oid}")
+
+    assert header["containerID"] == cid, "Tombstone Header CID is wrong"
+
+    with open(wallet_path, "r") as file:
+        wlt_data = json.loads(file.read())
+    wlt = wallet.Wallet.from_json(wlt_data, passwords=[WALLET_PASS])
+    addr = wlt.accounts[0].address
+
+    assert header["ownerID"] == addr, "Tombstone Owner ID is wrong"
+    assert header["objectType"] == "TOMBSTONE", "Header Type isn't Tombstone"
+    assert (
+        header["sessionToken"]["body"]["object"]["verb"] == "DELETE"
+    ), "Header Session Type isn't DELETE"
+    assert (
+        header["sessionToken"]["body"]["object"]["target"]["container"] == cid
+    ), "Header Session ID is wrong"
+    assert (
+        oid in header["sessionToken"]["body"]["object"]["target"]["objects"]
+    ), "Header Session OID is wrong"
+
+
+@allure.step("Delete Objects")
+def delete_objects(
+    storage_objects: list[StorageObjectInfo], shell: Shell, neofs_env: NeoFSEnv
+) -> None:
+    """
+    Deletes given storage objects.
+
+    Args:
+        storage_objects: list of objects to delete
+        shell: executor for cli command
+    """
+
+    with allure.step("Delete objects"):
+        for storage_object in storage_objects:
+            storage_object.tombstone = delete_object(
+                storage_object.wallet_file_path,
+                storage_object.cid,
+                storage_object.oid,
+                shell=shell,
+                endpoint=neofs_env.sn_rpc,
+            )
+            verify_head_tombstone(
+                wallet_path=storage_object.wallet_file_path,
+                cid=storage_object.cid,
+                oid_ts=storage_object.tombstone,
+                oid=storage_object.oid,
+                shell=shell,
+                endpoint=neofs_env.sn_rpc,
+            )
+
+    current_epoch = neofs_epoch.get_epoch(neofs_env)
+    neofs_epoch.tick_epoch(neofs_env)
+    neofs_epoch.wait_for_epochs_align(neofs_env, current_epoch)
+    sleep(CLEANUP_TIMEOUT)
+
+    with allure.step("Get objects and check errors"):
+        for storage_object in storage_objects:
+            with pytest.raises(Exception, match=OBJECT_ALREADY_REMOVED):
+                get_object(
+                    storage_object.wallet_file_path,
+                    storage_object.cid,
+                    storage_object.oid,
+                    shell=shell,
+                    endpoint=neofs_env.sn_rpc,
+                )