yuqei/ai-contract-platform
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
69c582892ff08819f58af610278bc0f3c10407a5
ai-contract-platform/yuqei-infra/tests/test_production_release_tools.py
T
yuqei 69c582892f
Build and Test Quality Gate / Contract Web typecheck and build (pull_request) Successful in 6m12s
Details
Build and Test Quality Gate / Python tests and Contract API migration (pull_request) Failing after 9m2s
Details
Security Quality Gate / npm audit, Bandit, pip-audit, detect-secrets (pull_request) Failing after 10m21s
Details
feat: harden identity diagnostics and SMTP delivery
2026-08-02 17:02:32 +08:00

628 lines
25 KiB
Python
Raw Blame History

from __future__ import annotations
import hashlib
import importlib.util
import io
import json
import sys
import tarfile
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from threading import Thread
from types import ModuleType
import pytest
REPO_ROOT = Path(__file__).resolve().parents[2]
def load_script(name: str, relative_path: str) -> ModuleType:
path = REPO_ROOT / relative_path
spec = importlib.util.spec_from_file_location(name, path)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
preflight = load_script("yuqei_production_preflight", "yuqei-infra/deploy/preflight-production.py")
backup_verify = load_script("yuqei_backup_verify", "yuqei-infra/backup/verify-production-backup.py")
provider_acceptance = load_script("yuqei_provider_acceptance", "yuqei-infra/scripts/run-ai-provider-acceptance.py")
def production_values() -> dict[str, str]:
values = preflight.parse_env_file(REPO_ROOT / "yuqei-infra/compose/.env.production.example")
replacements = {
"RELEASE_ID": "v2.0.0-rc.1",
"CONTRACT_API_IMAGE": f"registry.internal/contract-api@sha256:{'1' * 64}",
"AI_PLATFORM_API_IMAGE": f"registry.internal/ai-platform-api@sha256:{'2' * 64}",
"AI_WORKER_IMAGE": f"registry.internal/ai-worker@sha256:{'3' * 64}",
"POSTGRES_IMAGE": f"registry.internal/pgvector@sha256:{'4' * 64}",
"POSTGRES_CLIENT_IMAGE": f"registry.internal/postgres@sha256:{'5' * 64}",
"REDIS_IMAGE": f"registry.internal/redis@sha256:{'6' * 64}",
"MINIO_IMAGE": f"registry.internal/minio@sha256:{'7' * 64}",
"MINIO_MC_IMAGE": f"registry.internal/minio-mc@sha256:{'8' * 64}",
"NGINX_IMAGE": f"registry.internal/nginx@sha256:{'9' * 64}",
"BACKUP_UTILITY_IMAGE": f"registry.internal/alpine@sha256:{'a' * 64}",
"BACKEND_SERVER_NAME": "api.contracts.acme.cn",
"BACKEND_PUBLIC_BASE_URL": "https://api.contracts.acme.cn",
"POSTGRES_USER": "yuqei_prod",
"POSTGRES_PASSWORD": "postgres-secret-that-is-long-and-unique-001",
"MINIO_ROOT_USER": "yuqei_prod_minio",
"MINIO_ROOT_PASSWORD": "minio-secret-that-is-long-and-unique-002",
"CONTRACT_API_WEB_PROXY_SHARED_SECRET": "proxy-secret-that-is-long-and-unique-003",
"CONTRACT_API_IDENTITY_RATE_LIMIT_SECRET": "identity-secret-that-is-long-and-unique-004",
"CONTRACT_API_ANTI_ABUSE_HASH_SECRET": "contract-anti-abuse-secret-that-is-long-004a",
"CONTRACT_WEB_SESSION_SECRET": "session-secret-that-is-long-and-unique-005",
"AI_PLATFORM_SERVICE_SHARED_SECRET": "ai-service-secret-that-is-long-and-unique-006",
"AI_PLATFORM_ANTI_ABUSE_HASH_SECRET": "ai-anti-abuse-secret-that-is-long-and-unique-006a",
"CONTRACT_API_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_contract",
"AI_PLATFORM_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_ai",
"YUQEI_WORKER_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_ai",
}
values.update(replacements)
return values
def test_production_preflight_accepts_strict_configuration() -> None:
checks = preflight.validate_environment(production_values())
assert [check.key for check in checks if check.status == "fail"] == []
def test_disabled_identity_email_does_not_require_smtp_secret_file() -> None:
values = production_values()
values.pop("CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST", None)
environment_checks = {check.key: check for check in preflight.validate_environment(values)}
file_checks = {check.key: check for check in preflight.validate_files(values)}
assert environment_checks["identity_email_delivery"].status == "pass"
assert file_checks["smtp_password_file"].status == "pass"
def test_smtp_identity_email_requires_and_validates_opt_in_secret_file(tmp_path: Path) -> None:
password_file = tmp_path / "contract-smtp-password"
password_file.write_text("test-smtp-password\n", encoding="utf-8")
password_file.chmod(0o600)
values = production_values()
values.update(
{
"CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE": "smtp",
"CONTRACT_API_IDENTITY_SELF_SERVICE_ENABLED": "true",
"CONTRACT_API_IDENTITY_MANUAL_INVITATION_ENABLED": "false",
"CONTRACT_API_IDENTITY_SMTP_HOST": "smtp.mail.acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USERNAME": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD": "",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE": "/run/secrets/contract_smtp_password",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST": str(password_file.resolve()),
"CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS": "true",
"CONTRACT_API_IDENTITY_SMTP_USE_SSL": "false",
"CONTRACT_API_IDENTITY_PUBLIC_WEB_BASE_URL": "https://contracts.acme.cn",
}
)
environment_checks = {check.key: check for check in preflight.validate_environment(values)}
file_checks = {check.key: check for check in preflight.validate_files(values)}
assert environment_checks["identity_email_delivery"].status == "pass"
assert file_checks["smtp_password_file"].status == "pass"
if "smtp_password_permissions" in file_checks:
assert file_checks["smtp_password_permissions"].status == "pass"
unencrypted = dict(values)
unencrypted["CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS"] = "false"
unencrypted["CONTRACT_API_IDENTITY_SMTP_USE_SSL"] = "false"
unencrypted_checks = {
check.key: check for check in preflight.validate_environment(unencrypted)
}
assert unencrypted_checks["identity_email_delivery"].status == "fail"
password_file.unlink()
missing_file_checks = {check.key: check for check in preflight.validate_files(values)}
assert missing_file_checks["smtp_password_file"].status == "fail"
@pytest.mark.parametrize(
("key", "value"),
[
("CONTRACT_API_IDENTITY_SMTP_PORT", "0"),
("CONTRACT_API_IDENTITY_SMTP_PORT", "not-a-port"),
("CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS", "0"),
("CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS", "121"),
("CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL", "not-an-email"),
("CONTRACT_API_IDENTITY_SMTP_HOST", " smtp.mail.acme.cn"),
("CONTRACT_API_IDENTITY_SMTP_FROM_NAME", "AI Contract\r\nBcc: hidden@example.com"),
("CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS", "sometimes"),
],
)
def test_smtp_preflight_rejects_runtime_invalid_values(
tmp_path: Path,
key: str,
value: str,
) -> None:
password_file = tmp_path / "contract-smtp-password"
password_file.write_text("test-smtp-password\n", encoding="utf-8")
values = production_values()
values.update(
{
"CONTRACT_API_IDENTITY_EMAIL_DELIVERY_MODE": "smtp",
"CONTRACT_API_IDENTITY_SELF_SERVICE_ENABLED": "true",
"CONTRACT_API_IDENTITY_MANUAL_INVITATION_ENABLED": "false",
"CONTRACT_API_IDENTITY_SMTP_HOST": "smtp.mail.acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USERNAME": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD": "",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE": "/run/secrets/contract_smtp_password",
"CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST": str(password_file.resolve()),
"CONTRACT_API_IDENTITY_SMTP_FROM_EMAIL": "mailer@acme.cn",
"CONTRACT_API_IDENTITY_SMTP_USE_STARTTLS": "true",
"CONTRACT_API_IDENTITY_SMTP_USE_SSL": "false",
"CONTRACT_API_IDENTITY_SMTP_TIMEOUT_SECONDS": "10",
"CONTRACT_API_IDENTITY_PUBLIC_WEB_BASE_URL": "https://contracts.acme.cn",
key: value,
}
)
checks = {check.key: check for check in preflight.validate_environment(values)}
assert checks["identity_email_delivery"].status == "fail"
def test_smtp_production_override_mounts_only_the_password_file_secret() -> None:
override = (
REPO_ROOT / "yuqei-infra/compose/docker-compose.production.smtp.yml"
).read_text(encoding="utf-8")
assert "CONTRACT_API_IDENTITY_SMTP_PASSWORD: \"\"" in override
assert "CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE: /run/secrets/contract_smtp_password" in override
assert "target: contract_smtp_password" in override
assert "file: ${CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST:?" in override
def test_production_preflight_rejects_smoke_database_public_bind_and_fallback() -> None:
values = production_values()
values["CONTRACT_POSTGRES_DB"] = "yuqei_contract_smoke"
values["CONTRACT_API_DATABASE_URL"] = (
"postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_contract_smoke"
)
values["POSTGRES_BIND_ADDRESS"] = "0.0.0.0"
values["AI_PLATFORM_PROVIDER_FALLBACK_ENABLED"] = "true"
failed = {check.key for check in preflight.validate_environment(values) if check.status == "fail"}
assert {"production_modes", "internal_port_binding", "database_boundaries"} <= failed
def test_backup_manifest_and_contract_archive_are_verified(tmp_path: Path) -> None:
backup_dir = tmp_path / "backup-20260716T120000Z"
backup_dir.mkdir()
archive_path = backup_dir / "contract-files.tar.gz"
with tarfile.open(archive_path, mode="w:gz") as archive:
content = b"contract"
info = tarfile.TarInfo("contract-1/document.txt")
info.size = len(content)
archive.addfile(info, io.BytesIO(content))
dump_path = backup_dir / "postgres-contract.dump"
dump_path.write_bytes(b"postgres-dump")
artifacts = []
for path in (archive_path, dump_path):
artifacts.append(
{
"path": path.name,
"size_bytes": path.stat().st_size,
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
}
)
(backup_dir / "manifest.json").write_text(
json.dumps(
{
"schema_version": "yuqei-production-backup.v1",
"backup_id": backup_dir.name,
"created_at": "2026-07-16T12:00:00+00:00",
"artifacts": artifacts,
}
),
encoding="utf-8",
)
manifest, verified = backup_verify.validate_backup(backup_dir)
assert manifest["backup_id"] == backup_dir.name
assert len(verified) == 2
def test_provider_acceptance_separates_technical_success_from_legal_approval() -> None:
result = provider_acceptance.evaluate_case(
row={
"sample_id": "KBQA-1",
"expected_citation_required": "true",
"expected_source_ids": "LAW-CIVIL-001",
"production_ready": "false",
"status": "LEGAL_REVIEW",
"legal_reviewer": "待法务复核",
},
suite="legal_qa",
response={
"answer": "应结合合同和证据进一步判断。",
"citations": [{"document_id": "LAW-CIVIL-001"}],
},
audit={
"id": "run-1",
"status": "succeeded",
"provider_status": "succeeded",
"provider": "deepseek",
"model": "deepseek-chat",
"quality_labels": [],
},
latency_ms=100,
expected_boundary="不得直接判断个案效力。",
expected_severity=None,
)
assert result.technical_status == "passed"
assert result.legal_status == "needs_legal_review"
assert result.matched_source_ids == ["LAW-CIVIL-001"]
def test_provider_acceptance_rejects_fallback_audit() -> None:
result = provider_acceptance.evaluate_case(
row={"sample_id": "KBQA-2", "expected_citation_required": "false"},
suite="legal_qa",
response={"answer": "fallback answer", "citations": []},
audit={
"id": "run-2",
"status": "fallback",
"provider_status": "failed",
"provider_error": "provider unavailable",
"quality_labels": [{"code": "fallback_used"}],
},
latency_ms=50,
expected_boundary="",
expected_severity=None,
)
assert result.technical_status == "failed"
assert "provider_not_succeeded" in result.problems
assert "fallback_or_provider_failure_label" in result.problems
def test_provider_acceptance_requires_explicit_stable_source_ids() -> None:
observed = provider_acceptance._response_source_ids(
[
{"document_id": "doc-LAW-CIVIL-001-random"},
{"document_id": "doc-random", "reference": "LAW-CONTRACT-001 article 1"},
{"document_id": "doc-random-2", "source_ids": ["LAW-CIVIL-001", "not-stable"]},
]
)
assert observed == {"LAW-CIVIL-001", "LAW-CONTRACT-001"}
def test_provider_acceptance_loads_phase13_knowledge_cards_idempotently() -> None:
documents: list[dict[str, object]] = []
class Client:
def get(self, path: str):
assert path.startswith("/api/v1/knowledge-documents/search?")
return documents, 1
def post(self, path: str, payload: dict[str, object]):
assert path == "/api/v1/knowledge-documents"
documents.append({"id": f"doc-{len(documents)}", **payload})
return documents[-1], 1
client = Client()
assert provider_acceptance.load_phase13_knowledge_assets(client) == 10
assert provider_acceptance.load_phase13_knowledge_assets(client) == 0
assert len(documents) == 10
assert all(str(item["reference"]).startswith("LAW-") for item in documents)
assert all("不代表法务批准" in str(item["content"]) for item in documents)
def test_provider_acceptance_evidence_excludes_responses_and_identifiers(tmp_path: Path) -> None:
result = provider_acceptance.evaluate_case(
row={
"sample_id": "sensitive-sample-id",
"expected_citation_required": "true",
"expected_source_ids": "LAW-SECRET-001",
"production_ready": "false",
},
suite="legal_qa",
response={
"answer": "sensitive-provider-answer",
"citations": [{"document_id": "LAW-SECRET-001"}],
},
audit={
"id": "sensitive-audit-run-id",
"status": "succeeded",
"provider_status": "succeeded",
"provider": "deepseek",
"model": "deepseek-chat",
"quality_labels": [],
},
latency_ms=100,
expected_boundary="sensitive-legal-boundary",
expected_severity=None,
)
summary = provider_acceptance.build_evidence_summary(
status="needs_legal_review",
provider="deepseek",
model="deepseek-chat",
cases=[result],
)
provider_acceptance.write_evidence(tmp_path, summary)
assert set(summary) == {"provider", "model", "status", "counts", "problems"}
serialized = json.dumps(summary, ensure_ascii=False)
for sensitive_value in (
"sensitive-sample-id",
"sensitive-provider-answer",
"LAW-SECRET-001",
"sensitive-audit-run-id",
"sensitive-legal-boundary",
):
assert sensitive_value not in serialized
assert not (tmp_path / "ai-provider-acceptance-responses.json").exists()
assert set(path.name for path in tmp_path.iterdir()) == {
"ai-provider-acceptance-summary.json",
"ai-provider-acceptance-summary.md",
}
def test_provider_acceptance_validates_safe_runtime_parameters() -> None:
valid_key = "service-key-that-is-long-enough-for-testing"
with pytest.raises(ValueError, match="query/fragment"):
provider_acceptance.ApiClient(
"https://api.example.test/ai?key=unsafe",
service_key=valid_key,
tenant_id="default",
timeout_seconds=30,
)
with pytest.raises(ValueError, match="service key"):
provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key=f"{valid_key}\r\nInjected: true",
tenant_id="default",
timeout_seconds=30,
)
with pytest.raises(ValueError, match="Tenant ID"):
provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key=valid_key,
tenant_id="default\r\nInjected: true",
timeout_seconds=30,
)
with pytest.raises(ValueError, match="Timeout"):
provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key=valid_key,
tenant_id="default",
timeout_seconds=0,
)
def test_provider_acceptance_allows_explicit_isolated_memory_schema_warning() -> None:
class Client:
def get(self, path: str):
if path == "/api/v1/runtime/schema-check":
return {
"status": "warning",
"checks": [
{
"name": "repository_backend",
"status": "warning",
"details": {"repository_backend": "memory"},
},
{"name": "anti_abuse_runtime", "status": "warning", "details": {}},
],
}, 1
if path == "/api/v1/model-configs":
return [
{
"provider": "isolated-provider",
"model": "isolated-model",
"enabled": True,
"is_default": True,
"base_url": "https://provider.example.test",
"api_key_set": True,
}
], 1
raise AssertionError(path)
schema, provider_config = provider_acceptance.validate_runtime(Client())
assert schema["status"] == "warning"
assert provider_config["provider"] == "isolated-provider"
def test_provider_acceptance_encodes_chinese_request_as_utf8(monkeypatch) -> None:
captured: dict[str, object] = {}
class Response:
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def read(self, _limit: int) -> bytes:
return b'{"status":"ok"}'
def fake_urlopen(request, *, timeout):
captured["body"] = request.data
captured["timeout"] = timeout
return Response()
monkeypatch.setattr(provider_acceptance, "urlopen", fake_urlopen)
client = provider_acceptance.ApiClient(
"https://api.example.test/ai",
service_key="service-key-that-is-long-enough-for-testing",
tenant_id="default",
timeout_seconds=30,
)
payload, _ = client.post("/api/v1/legal/qa", {"question": "电子邮件能否作为合同证据？"})
assert payload == {"status": "ok"}
assert json.loads(captured["body"].decode("utf-8"))["question"] == "电子邮件能否作为合同证据？"
assert captured["timeout"] == 30
def test_provider_acceptance_runs_all_phase13_cases_against_isolated_http_service() -> None:
qa_rows = provider_acceptance.load_rows(provider_acceptance.QA_INVENTORY)
review_rows = provider_acceptance.load_rows(provider_acceptance.REVIEW_INVENTORY)
qa_by_question = {row["question"]: row for row in qa_rows}
review_by_id = {row["sample_id"]: row for row in review_rows}
audits: dict[str, dict[str, object]] = {}
class Handler(BaseHTTPRequestHandler):
def send_json(self, payload: object) -> None:
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None: # noqa: N802
path = self.path.split("?", 1)[0]
if path == "/api/v1/runtime/schema-check":
self.send_json(
{
"status": "ok",
"checks": [{"name": "provider_runtime", "status": "ok"}],
}
)
return
if path == "/api/v1/model-configs":
self.send_json(
[
{
"id": "provider-isolated",
"provider": "isolated-provider",
"model": "isolated-model",
"enabled": True,
"is_default": True,
"base_url": "https://provider.invalid/v1",
"api_key_set": True,
}
]
)
return
if path == "/api/v1/audit/ai-runs":
self.send_json(list(reversed(list(audits.values()))))
return
run_id = path.removeprefix("/api/v1/audit/ai-runs/")
if run_id in audits:
self.send_json(audits[run_id])
return
self.send_error(404)
def do_POST(self) -> None: # noqa: N802
content_length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(content_length).decode("utf-8"))
run_id = f"isolated-run-{len(audits) + 1}"
if self.path == "/api/v1/legal/qa":
row = qa_by_question[payload["question"]]
citations = [
{"document_id": f"doc-{index}", "source_ids": [source_id]}
for index, source_id in enumerate(provider_acceptance.split_ids(row["expected_source_ids"]))
]
operation = "legal_qa"
response = {"answer": "基于检索依据给出边界明确的答复。", "citations": citations}
elif self.path == "/api/v1/legal/contract-review":
row = review_by_id[payload["contract_title"]]
citations = [
{"document_id": f"doc-{index}", "source_ids": [source_id]}
for index, source_id in enumerate(provider_acceptance.split_ids(row["expected_source_citations"]))
]
cited_source_ids = provider_acceptance.split_ids(row["expected_source_citations"])
operation = "contract_review"
response = {
"summary": "已按约定审查重点完成审查。",
"citations": citations,
"risk_items": [
{
"severity": row["expected_severity"],
"citation_source_ids": cited_source_ids,
}
],
"audit_run_id": run_id,
}
else:
self.send_error(404)
return
audits[run_id] = {
"id": run_id,
"operation": operation,
"status": "succeeded",
"provider_status": "succeeded",
"provider": "isolated-provider",
"model": "isolated-model",
"quality_labels": [],
"created_at": datetime.now(timezone.utc).isoformat(),
}
self.send_json(response)
def log_message(self, _format: str, *_args: object) -> None:
return
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
client = provider_acceptance.ApiClient(
f"http://127.0.0.1:{server.server_port}",
service_key="isolated-non-secret-service-key-for-tests",
tenant_id="default",
timeout_seconds=10,
)
schema, provider_config = provider_acceptance.validate_runtime(client)
seen_run_ids: set[str] = set()
results = [
provider_acceptance.run_qa_case(client, row, seen_run_ids)[0]
for row in qa_rows
]
results.extend(
provider_acceptance.run_review_case(client, row, seen_run_ids)[0]
for row in review_rows
)
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
assert len(qa_rows) == 10
assert len(review_rows) == 8
assert len(results) == 18
assert schema["status"] == "ok"
assert provider_config["provider"] == "isolated-provider"
assert all(result.technical_status == "passed" for result in results)
assert all(result.legal_status == "needs_legal_review" for result in results)
def test_production_tool_bandit_suppressions_are_exactly_allowlisted() -> None:
expected = {
"yuqei-infra/backup/run-production-backup.py": ["B404", "B603"],
"yuqei-infra/backup/verify-production-backup.py": ["B404", "B603", "B603"],
"yuqei-infra/observability/check-production-health.py": ["B310", "B310"],
"yuqei-infra/scripts/run-ai-provider-acceptance.py": ["B310"],
}
observed: dict[str, list[str]] = {}
for relative_path in expected:
lines = (REPO_ROOT / relative_path).read_text(encoding="utf-8").splitlines()
observed[relative_path] = [
line.split("# nosec", 1)[1].strip()
for line in lines
if "# nosec" in line
]
assert observed == expected
Reference in New Issue View Git Blame Copy Permalink