99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Create a one-user SQLite handoff without exposing the legacy database to HLCC."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import pathlib
|
||
|
|
import pwd
|
||
|
|
import sqlite3
|
||
|
|
import tempfile
|
||
|
|
|
||
|
|
|
||
|
|
OWNER_NAME = "bingshuo"
|
||
|
|
|
||
|
|
|
||
|
|
def prepare(source: pathlib.Path, destination: pathlib.Path, owner: str) -> None:
|
||
|
|
if not source.is_file():
|
||
|
|
raise RuntimeError("legacy database unavailable")
|
||
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
legacy = sqlite3.connect(f"file:{source}?mode=ro", uri=True, timeout=15)
|
||
|
|
try:
|
||
|
|
schema = legacy.execute(
|
||
|
|
"select sql from sqlite_master where type = 'table' and name = 'user'"
|
||
|
|
).fetchone()
|
||
|
|
row = legacy.execute(
|
||
|
|
"select * from user where lower_name = ?",
|
||
|
|
(OWNER_NAME,),
|
||
|
|
).fetchall()
|
||
|
|
if not schema or not schema[0] or len(row) != 1:
|
||
|
|
raise RuntimeError("legacy owner identity mismatch")
|
||
|
|
|
||
|
|
file_descriptor, temporary_name = tempfile.mkstemp(
|
||
|
|
prefix=".owner-identity-source.",
|
||
|
|
suffix=".db",
|
||
|
|
dir=destination.parent,
|
||
|
|
)
|
||
|
|
os.close(file_descriptor)
|
||
|
|
temporary = pathlib.Path(temporary_name)
|
||
|
|
try:
|
||
|
|
handoff = sqlite3.connect(temporary)
|
||
|
|
try:
|
||
|
|
handoff.execute(schema[0])
|
||
|
|
columns = [item[1] for item in legacy.execute("pragma table_info(user)")]
|
||
|
|
placeholders = ",".join("?" for _column in columns)
|
||
|
|
column_sql = ",".join(f'"{column}"' for column in columns)
|
||
|
|
handoff.execute(
|
||
|
|
f"insert into user ({column_sql}) values ({placeholders})",
|
||
|
|
row[0],
|
||
|
|
)
|
||
|
|
handoff.commit()
|
||
|
|
finally:
|
||
|
|
handoff.close()
|
||
|
|
temporary.chmod(0o600)
|
||
|
|
identity = pwd.getpwnam(owner)
|
||
|
|
os.chown(temporary, identity.pw_uid, identity.pw_gid)
|
||
|
|
temporary.replace(destination)
|
||
|
|
finally:
|
||
|
|
temporary.unlink(missing_ok=True)
|
||
|
|
finally:
|
||
|
|
legacy.close()
|
||
|
|
|
||
|
|
receipt = destination.with_suffix(".receipt.json")
|
||
|
|
receipt.write_text(
|
||
|
|
json.dumps(
|
||
|
|
{
|
||
|
|
"schema": "guanghu.hlcc-owner-identity-handoff/v1",
|
||
|
|
"owner": OWNER_NAME,
|
||
|
|
"rows": 1,
|
||
|
|
"contains_repository_data": False,
|
||
|
|
"contains_access_tokens": False,
|
||
|
|
"result": "PREPARED",
|
||
|
|
},
|
||
|
|
ensure_ascii=False,
|
||
|
|
indent=2,
|
||
|
|
)
|
||
|
|
+ "\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
identity = pwd.getpwnam(owner)
|
||
|
|
os.chown(receipt, identity.pw_uid, identity.pw_gid)
|
||
|
|
receipt.chmod(0o600)
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--source", required=True, type=pathlib.Path)
|
||
|
|
parser.add_argument("--destination", required=True, type=pathlib.Path)
|
||
|
|
parser.add_argument("--owner", default="guanghu")
|
||
|
|
arguments = parser.parse_args()
|
||
|
|
prepare(arguments.source, arguments.destination, arguments.owner)
|
||
|
|
print("OWNER_IDENTITY_SOURCE_PREPARED rows=1 tokens=0 repositories=0")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|
||
|
|
|