Spaces:
Running
Running
File size: 12,717 Bytes
4d56236 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 | """Contract engine for managing contract lifecycle and operations."""
from datetime import datetime
from typing import List, Optional
from uuid import UUID
from sqlmodel import Session, create_engine
from fastmcp.utilities.logging import get_logger
from .contract import Contract, ContractState, ContractCreateRequest, ContractProposeRequest, ContractSignRequest, ContractRevokeRequest
from .crypto import ContractSigner, CryptoError
from .registry import ContractRegistry
logger = get_logger(__name__)
class ContractEngine:
"""Engine for managing contract lifecycle and operations."""
def __init__(self, database_url: str = "sqlite:///contracts.db"):
"""Initialize contract engine with database connection.
Args:
database_url: Database connection URL
"""
self.engine = create_engine(database_url, echo=False)
self._create_tables()
def _create_tables(self):
"""Create database tables."""
try:
Contract.metadata.create_all(self.engine)
logger.info("Contract database tables created/verified")
except Exception as e:
logger.error(f"Failed to create contract tables: {e}")
raise
def get_session(self) -> Session:
"""Get database session.
Returns:
SQLModel database session
"""
return Session(self.engine)
def get_registry(self) -> ContractRegistry:
"""Get contract registry.
Returns:
Contract registry instance
"""
return ContractRegistry(self.get_session())
async def create_contract(self, request: ContractCreateRequest, created_by: str) -> Contract:
"""Create a new contract.
Args:
request: Contract creation request
created_by: ID of the creating party
Returns:
Created contract instance
Raises:
ValueError: If contract creation fails
"""
try:
registry = self.get_registry()
# Convert request to contract data
contract_data = {
"title": request.title,
"description": request.description,
"clauses": [clause.model_dump() for clause in request.clauses],
"parties": request.parties,
"is_hipaa_compliant": request.is_hipaa_compliant,
"hipaa_entities": request.hipaa_entities or [],
"expires_at": request.expires_at,
"metadata": request.metadata,
"version": request.version
}
contract = registry.create_contract(contract_data, created_by)
logger.info(f"Created contract {contract.id} by {created_by}")
return contract
except Exception as e:
logger.error(f"Failed to create contract: {e}")
raise
async def get_contract(self, contract_id: UUID) -> Optional[Contract]:
"""Get a contract by ID.
Args:
contract_id: Contract UUID
Returns:
Contract instance or None if not found
"""
try:
registry = self.get_registry()
return registry.get_contract(contract_id)
except Exception as e:
logger.error(f"Failed to get contract {contract_id}: {e}")
return None
async def list_contracts(self, state: Optional[ContractState] = None,
created_by: Optional[str] = None) -> List[Contract]:
"""List contracts with optional filtering.
Args:
state: Optional state filter
created_by: Optional creator filter
Returns:
List of contract instances
"""
try:
registry = self.get_registry()
return registry.list_contracts(state, created_by)
except Exception as e:
logger.error(f"Failed to list contracts: {e}")
return []
async def propose_contract(self, contract_id: UUID, request: ContractProposeRequest,
proposed_by: str) -> Optional[Contract]:
"""Propose a contract to parties.
Args:
contract_id: Contract UUID
request: Proposal request
proposed_by: ID of the proposing party
Returns:
Updated contract instance or None if not found
Raises:
ValueError: If proposal fails
"""
try:
registry = self.get_registry()
# Validate contract exists and can be proposed
contract = registry.get_contract(contract_id)
if not contract:
raise ValueError("Contract not found")
if contract.state != ContractState.DRAFT:
raise ValueError(f"Cannot propose contract in state {contract.state}")
# Update metadata with proposal info
metadata = {
"proposal": {
"proposed_to": request.proposed_to,
"message": request.message,
"proposed_by": proposed_by,
"timestamp": datetime.utcnow().isoformat()
}
}
# Update state to proposed
updated_contract = registry.update_contract_state(
contract_id, ContractState.PROPOSED, proposed_by, metadata
)
logger.info(f"Proposed contract {contract_id} to {request.proposed_to} by {proposed_by}")
return updated_contract
except Exception as e:
logger.error(f"Failed to propose contract {contract_id}: {e}")
raise
async def sign_contract(self, contract_id: UUID, request: ContractSignRequest) -> Optional[Contract]:
"""Sign a contract.
Args:
contract_id: Contract UUID
request: Signing request
Returns:
Updated contract instance or None if not found
Raises:
ValueError: If signing fails
"""
try:
registry = self.get_registry()
# Validate contract exists and can be signed
contract = registry.get_contract(contract_id)
if not contract:
raise ValueError("Contract not found")
if contract.state not in [ContractState.PROPOSED, ContractState.SIGNED]:
raise ValueError(f"Cannot sign contract in state {contract.state}")
# Check if party is already signed
existing_signatures = contract.get_signatures()
if any(sig.signer_id == request.signer_id for sig in existing_signatures):
raise ValueError("Party has already signed this contract")
# Verify signature
if not self._verify_contract_signature(contract, request):
raise ValueError("Invalid signature")
# Create signature object
from .contract import Signature
signature = Signature(
signer_id=request.signer_id,
signer_type=request.signer_type,
signature=request.signature,
public_key=request.public_key,
metadata=request.metadata
)
# Add signature
updated_contract = registry.add_signature(contract_id, signature.model_dump())
logger.info(f"Signed contract {contract_id} by {request.signer_id}")
return updated_contract
except Exception as e:
logger.error(f"Failed to sign contract {contract_id}: {e}")
raise
async def revoke_contract(self, contract_id: UUID, request: ContractRevokeRequest) -> Optional[Contract]:
"""Revoke a contract.
Args:
contract_id: Contract UUID
request: Revocation request
Returns:
Updated contract instance or None if not found
Raises:
ValueError: If revocation fails
"""
try:
registry = self.get_registry()
# Validate contract exists and can be revoked
contract = registry.get_contract(contract_id)
if not contract:
raise ValueError("Contract not found")
if contract.state == ContractState.REVOKED:
raise ValueError("Contract is already revoked")
if contract.state == ContractState.EXPIRED:
raise ValueError("Cannot revoke expired contract")
# Revoke contract
updated_contract = registry.revoke_contract(
contract_id, request.reason, request.revoked_by, request.metadata
)
logger.info(f"Revoked contract {contract_id} by {request.revoked_by}: {request.reason}")
return updated_contract
except Exception as e:
logger.error(f"Failed to revoke contract {contract_id}: {e}")
raise
def _verify_contract_signature(self, contract: Contract, request: ContractSignRequest) -> bool:
"""Verify a contract signature.
Args:
contract: Contract instance
request: Signing request
Returns:
True if signature is valid, False otherwise
"""
try:
from .crypto import verify_signature
# Create signing message
signing_message = f"{contract.id}:{contract.get_content_hash()}:{request.signer_id}:{request.signer_type}"
# Verify signature
return verify_signature(
request.public_key,
signing_message,
request.signature
)
except Exception as e:
logger.error(f"Signature verification failed: {e}")
return False
async def get_contracts_by_party(self, party_id: str) -> List[Contract]:
"""Get contracts involving a specific party.
Args:
party_id: Party ID to search for
Returns:
List of contracts involving the party
"""
try:
registry = self.get_registry()
return registry.get_contracts_by_party(party_id)
except Exception as e:
logger.error(f"Failed to get contracts for party {party_id}: {e}")
return []
async def cleanup_expired_contracts(self) -> int:
"""Mark expired contracts as expired.
Returns:
Number of contracts marked as expired
"""
try:
registry = self.get_registry()
count = registry.mark_contracts_expired()
logger.info(f"Marked {count} contracts as expired")
return count
except Exception as e:
logger.error(f"Failed to cleanup expired contracts: {e}")
return 0
async def get_contract_statistics(self) -> dict:
"""Get contract statistics.
Returns:
Dictionary with contract statistics
"""
try:
registry = self.get_registry()
# Get all contracts
all_contracts = registry.list_contracts()
# Count by state
state_counts = {}
for state in ContractState:
state_counts[state.value] = len([c for c in all_contracts if c.state == state])
# Count HIPAA contracts
hipaa_count = len([c for c in all_contracts if c.is_hipaa_compliant])
# Count signed contracts
signed_count = len([c for c in all_contracts if c.state == ContractState.SIGNED])
return {
"total_contracts": len(all_contracts),
"by_state": state_counts,
"hipaa_compliant": hipaa_count,
"signed_contracts": signed_count,
"expired_contracts": len(registry.get_expired_contracts())
}
except Exception as e:
logger.error(f"Failed to get contract statistics: {e}")
return {}
|