Spaces:

UNI12345
/

ARCHIVISE

Sleeping

ARCHIVISE / app /utils /github_client.py

Initialize Archvise API backend for Hugging Face Spaces

50925ca 4 months ago

6.46 kB

import base64
import hashlib
import re
from typing import List, Dict, Any
import httpx
from github import Github
from cryptography.fernet import Fernet
from app.config import settings
from loguru import logger

# Initialize encryption cipher using a key derived from SECRET_KEY
def _get_fernet_cipher() -> Fernet:
    # Fernet requires a 32-byte base64-encoded key
    key_hash = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
    fernet_key = base64.urlsafe_b64encode(key_hash)
    return Fernet(fernet_key)

def encrypt_token(token: str) -> str:
    cipher = _get_fernet_cipher()
    return cipher.encrypt(token.encode()).decode()

def decrypt_token(encrypted_token: str) -> str:
    cipher = _get_fernet_cipher()
    return cipher.decrypt(encrypted_token.encode()).decode()

def get_github_auth_url(state: str) -> str:
    redirect_uri = f"{settings.FRONTEND_URL}/api/auth/callback"
    return (
        f"https://github.com/login/oauth/authorize"
        f"?client_id={settings.GITHUB_CLIENT_ID}"
        f"&redirect_uri={redirect_uri}"
        f"&state={state}"
        f"&scope=repo,read:org"
    )

async def exchange_code_for_token(code: str) -> dict:
    url = "https://github.com/login/oauth/access_token"
    headers = {"Accept": "application/json"}
    data = {
        "client_id": settings.GITHUB_CLIENT_ID,
        "client_secret": settings.GITHUB_CLIENT_SECRET,
        "code": code
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, headers=headers, data=data)
        response.raise_for_status()
        return response.json()

async def get_github_user_info(access_token: str) -> dict:
    url = "https://api.github.com/user"
    headers = {
        "Authorization": f"token {access_token}",
        "Accept": "application/vnd.github.v3+json"
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(url, headers=headers)
        response.raise_for_status()
        return response.json()

def get_user_repos(access_token: str) -> List[Dict[str, Any]]:
    try:
        g = Github(access_token)
        repos = []
        # Get repos sorted by updated date
        for repo in g.get_user().get_repos(sort="updated", direction="desc"):
            repos.append({
                "id": repo.id,
                "name": repo.name,
                "full_name": repo.full_name,
                "html_url": repo.html_url,
                "description": repo.description,
                "language": repo.language,
                "updated_at": repo.updated_at.isoformat() if repo.updated_at else "",
                "private": repo.private
            })
        return repos
    except Exception as e:
        logger.error(f"Failed to fetch repos from GitHub: {e}")
        return []

def get_repo_branches(access_token: str, repo_full_name: str) -> List[str]:
    try:
        g = Github(access_token)
        repo = g.get_repo(repo_full_name)
        return [b.name for b in repo.get_branches()]
    except Exception as e:
        logger.error(f"Failed to fetch branches for {repo_full_name}: {e}")
        return []

# File Extensions Whitelist
ALLOWED_EXTENSIONS = {
    ".py", ".js", ".ts", ".jsx", ".tsx", ".json", ".yaml", ".yml",
    ".toml", ".html", ".css", ".sql", ".go", ".java", ".rb"
}

def get_repo_files(access_token: str, repo_full_name: str, branch: str) -> List[Dict[str, str]]:
    try:
        g = Github(access_token)
        repo = g.get_repo(repo_full_name)
        
        # Get files recursively up to limits
        files = []
        total_size = 0
        max_size = 50 * 1024 * 1024  # 50MB limit
        
        # Get git tree recursively
        git_ref = repo.get_git_ref(f"heads/{branch}")
        tree = repo.get_git_tree(git_ref.object.sha, recursive=True)
        
        for element in tree.tree:
            if element.type == "blob":
                # Check extension
                import os
                _, ext = os.path.splitext(element.path)
                if ext.lower() in ALLOWED_EXTENSIONS:
                    # Ignore standard vendor/lock folders to stay within context limits
                    if any(p in element.path.split("/") for p in ["node_modules", "vendor", "dist", "build", ".git", "venv", ".venv"]):
                        continue
                        
                    # Fetch blob content
                    blob = repo.get_git_blob(element.sha)
                    content = base64.b64decode(blob.content).decode("utf-8", errors="ignore")
                    
                    # Redact secrets
                    content = redact_env_secrets(content, element.path)
                    
                    files.append({
                        "filename": element.path,
                        "content": content
                    })
                    
                    total_size += len(content)
                    if len(files) >= 20 or total_size >= max_size:
                        break
                        
        return files
    except Exception as e:
        logger.error(f"Failed to fetch files from repo {repo_full_name}: {e}")
        return []

def redact_env_secrets(content: str, filename: str) -> str:
    # Redact common secrets patterns
    lines = content.splitlines()
    redacted_lines = []
    
    # Check if this is an env file
    is_env = filename.endswith(".env") or filename.endswith(".env.production") or filename.endswith(".env.local")
    
    # Generic credential keys regex
    secret_key_re = re.compile(
        r"(password|secret|key|token|auth|pwd|credential|private_key|api_key|access_key)\s*[:=]\s*['\"]?([a-zA-Z0-9_\-\.\=\+]{8,})['\"]?",
        re.IGNORECASE
    )
    
    for line in lines:
        if is_env:
            # For env files, redact everything after '='
            if "=" in line and not line.strip().startswith("#"):
                key, val = line.split("=", 1)
                redacted_lines.append(f"{key}=[REDACTED]")
            else:
                redacted_lines.append(line)
        else:
            # Inline check for code files
            match = secret_key_re.search(line)
            if match:
                # Replace matching group containing key value
                val = match.group(2)
                redacted_lines.append(line.replace(val, "[REDACTED]"))
            else:
                redacted_lines.append(line)
                
    return "\n".join(redacted_lines)

	import base64
	import hashlib
	import re
	from typing import List, Dict, Any
	import httpx
	from github import Github
	from cryptography.fernet import Fernet
	from app.config import settings
	from loguru import logger

	# Initialize encryption cipher using a key derived from SECRET_KEY
	def _get_fernet_cipher() -> Fernet:
	# Fernet requires a 32-byte base64-encoded key
	key_hash = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
	fernet_key = base64.urlsafe_b64encode(key_hash)
	return Fernet(fernet_key)

	def encrypt_token(token: str) -> str:
	cipher = _get_fernet_cipher()
	return cipher.encrypt(token.encode()).decode()

	def decrypt_token(encrypted_token: str) -> str:
	cipher = _get_fernet_cipher()
	return cipher.decrypt(encrypted_token.encode()).decode()

	def get_github_auth_url(state: str) -> str:
	redirect_uri = f"{settings.FRONTEND_URL}/api/auth/callback"
	return (
	f"https://github.com/login/oauth/authorize"
	f"?client_id={settings.GITHUB_CLIENT_ID}"
	f"&redirect_uri={redirect_uri}"
	f"&state={state}"
	f"&scope=repo,read:org"
	)

	async def exchange_code_for_token(code: str) -> dict:
	url = "https://github.com/login/oauth/access_token"
	headers = {"Accept": "application/json"}
	data = {
	"client_id": settings.GITHUB_CLIENT_ID,
	"client_secret": settings.GITHUB_CLIENT_SECRET,
	"code": code
	}

	async with httpx.AsyncClient() as client:
	response = await client.post(url, headers=headers, data=data)
	response.raise_for_status()
	return response.json()

	async def get_github_user_info(access_token: str) -> dict:
	url = "https://api.github.com/user"
	headers = {
	"Authorization": f"token {access_token}",
	"Accept": "application/vnd.github.v3+json"
	}
	async with httpx.AsyncClient() as client:
	response = await client.get(url, headers=headers)
	response.raise_for_status()
	return response.json()

	def get_user_repos(access_token: str) -> List[Dict[str, Any]]:
	try:
	g = Github(access_token)
	repos = []
	# Get repos sorted by updated date
	for repo in g.get_user().get_repos(sort="updated", direction="desc"):
	repos.append({
	"id": repo.id,
	"name": repo.name,
	"full_name": repo.full_name,
	"html_url": repo.html_url,
	"description": repo.description,
	"language": repo.language,
	"updated_at": repo.updated_at.isoformat() if repo.updated_at else "",
	"private": repo.private
	})
	return repos
	except Exception as e:
	logger.error(f"Failed to fetch repos from GitHub: {e}")
	return []

	def get_repo_branches(access_token: str, repo_full_name: str) -> List[str]:
	try:
	g = Github(access_token)
	repo = g.get_repo(repo_full_name)
	return [b.name for b in repo.get_branches()]
	except Exception as e:
	logger.error(f"Failed to fetch branches for {repo_full_name}: {e}")
	return []

	# File Extensions Whitelist
	ALLOWED_EXTENSIONS = {
	".py", ".js", ".ts", ".jsx", ".tsx", ".json", ".yaml", ".yml",
	".toml", ".html", ".css", ".sql", ".go", ".java", ".rb"
	}

	def get_repo_files(access_token: str, repo_full_name: str, branch: str) -> List[Dict[str, str]]:
	try:
	g = Github(access_token)
	repo = g.get_repo(repo_full_name)

	# Get files recursively up to limits
	files = []
	total_size = 0
	max_size = 50 * 1024 * 1024 # 50MB limit

	# Get git tree recursively
	git_ref = repo.get_git_ref(f"heads/{branch}")
	tree = repo.get_git_tree(git_ref.object.sha, recursive=True)

	for element in tree.tree:
	if element.type == "blob":
	# Check extension
	import os
	_, ext = os.path.splitext(element.path)
	if ext.lower() in ALLOWED_EXTENSIONS:
	# Ignore standard vendor/lock folders to stay within context limits
	if any(p in element.path.split("/") for p in ["node_modules", "vendor", "dist", "build", ".git", "venv", ".venv"]):
	continue

	# Fetch blob content
	blob = repo.get_git_blob(element.sha)
	content = base64.b64decode(blob.content).decode("utf-8", errors="ignore")

	# Redact secrets
	content = redact_env_secrets(content, element.path)

	files.append({
	"filename": element.path,
	"content": content
	})

	total_size += len(content)
	if len(files) >= 20 or total_size >= max_size:
	break

	return files
	except Exception as e:
	logger.error(f"Failed to fetch files from repo {repo_full_name}: {e}")
	return []

	def redact_env_secrets(content: str, filename: str) -> str:
	# Redact common secrets patterns
	lines = content.splitlines()
	redacted_lines = []

	# Check if this is an env file
	is_env = filename.endswith(".env") or filename.endswith(".env.production") or filename.endswith(".env.local")

	# Generic credential keys regex
	secret_key_re = re.compile(
	r"(password\|secret\|key\|token\|auth\|pwd\|credential\|private_key\|api_key\|access_key)\s[:=]\s['\"]?([a-zA-Z0-9_\-\.\=\+]{8,})['\"]?",
	re.IGNORECASE
	)

	for line in lines:
	if is_env:
	# For env files, redact everything after '='
	if "=" in line and not line.strip().startswith("#"):
	key, val = line.split("=", 1)
	redacted_lines.append(f"{key}=[REDACTED]")
	else:
	redacted_lines.append(line)
	else:
	# Inline check for code files
	match = secret_key_re.search(line)
	if match:
	# Replace matching group containing key value
	val = match.group(2)
	redacted_lines.append(line.replace(val, "[REDACTED]"))
	else:
	redacted_lines.append(line)

	return "\n".join(redacted_lines)