~drizzle-trunk/drizzle/development

CSVector *CSS3Protocol::s3_send(CSInputStream *input, const char *bucket, const char *key, off64_t size, const char *content_type, Md5Digest *digest, const char *s3Authorization, time_t s3AuthorizationTime)

1147

{

1148

CSStringBuffer *s3_buffer;

1149

char date[64];

1150

CSString *signed_str;

1151

uint32_t retry_count = 0;

1152

S3ProtocolCon *con_data;

1153

CSVector *replyHeaders;

1154

char checksum[32], *md5 = NULL;

1155

1156

enter_();

1157

push_(input);

1158

1159

new_(s3_buffer, CSStringBuffer());

1160

push_(s3_buffer);

1161

1162

new_(con_data, S3ProtocolCon());

1163

push_(con_data);

1164

1165

if (!content_type)

1166

content_type = "binary/octet-stream";

1167

1168

retry:

1169

1170

// Clear old settings.

1171

con_data->ms_reset();

1172

1173

if (s3Authorization) {

1174

SET_DATE_FROM_TIME(s3AuthorizationTime, date);

1175

} else {

1176

SET_DATE(date);

1177

}

1178

1179

// Build the URL

1180

s3_buffer->setLength(0);

1181

s3_buffer->append("http://");

1182

s3_buffer->append(bucket);

1183

s3_buffer->append(".");

1184

s3_buffer->append(s3_server->getCString());

1185

s3_buffer->append(key);

1186

1187

con_data->ms_setURL(s3_buffer->getCString());

1188

1189

// Add the 'DATE' header

1190

s3_buffer->setLength(0);

1191

s3_buffer->append("Date: ");

1192

s3_buffer->append(date);

1193

con_data->ms_setHeader(s3_buffer->getCString());

1194

1195

// Add the 'Content-Type' header

1196

s3_buffer->setLength(0);

1197

s3_buffer->append("Content-Type: ");

1198

s3_buffer->append(content_type);

1199

con_data->ms_setHeader(s3_buffer->getCString());

1200

1201

if (digest) {

1202

// Add the Md5 checksum header

1203

md5 = checksum;

1204

memset(checksum, 0, 32);

1205

base64Encode(digest->val, 16, checksum, 32);

1206

1207

s3_buffer->setLength(0);

1208

s3_buffer->append("Content-MD5: ");

1209

s3_buffer->append(checksum);

1210

con_data->ms_setHeader(s3_buffer->getCString());

1211

con_data->ms_calculate_md5 = false;

1212

} else

1213

con_data->ms_calculate_md5 = true;

1214

1215

1216

// Create the authentication signature and add the 'Authorization' header

1217

if (!s3Authorization)

1218

signed_str = s3_getSignature("PUT", md5, content_type, date, bucket, key);

1219

else

1220

signed_str = CSString::newString(s3Authorization);

1221

push_(signed_str);

1222

s3_buffer->setLength(0);

1223

s3_buffer->append("Authorization: AWS ");

1224

s3_buffer->append(s3_public_key->getCString());

1225

s3_buffer->append(":");

1226

s3_buffer->append(signed_str->getCString());

1227

release_(signed_str); signed_str = NULL;

1228

con_data->ms_setHeader(s3_buffer->getCString());

1229

1230

con_data->ms_execute_put_request(RETAIN(input), size);

1231

1232

if (con_data->ms_retry) {

1233

if (retry_count == s3_maxRetries) {

1234

CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 operation aborted after max retries.");

1235

}

1236

//printf("RETRY: s3_send()\n");

1237

retry_count++;

1238

input->reset();

1239

self->sleep(s3_sleepTime);

1240

goto retry;

1241

}

1242

1243

replyHeaders = con_data->ms_reply_headers.takeHeaders();

1244

1245

release_(con_data);

1246

release_(s3_buffer);

1247

release_(input);

1248

return_(replyHeaders);

1249

}

1250

1251

//-------------------------------

1252

CSString *CSS3Protocol::s3_getDataURL(const char *bucket, const char *key, uint32_t keep_alive)

1253

{

1254

CSStringBuffer *s3_buffer;

1255

char timeout[32];

1256

CSString *signed_str;

1257

enter_();

1258

1259

new_(s3_buffer, CSStringBuffer());

1260

push_(s3_buffer);

1261

1262

snprintf(timeout, 32, "%"PRId32"", ((uint32_t)time(NULL)) + keep_alive);

1263

1264

signed_str = s3_getSignature("GET", NULL, NULL, timeout, bucket, key);

1265

//printf("Unsafe: \"%s\"\n", signed_str->getCString());

1266

signed_str = urlEncode(signed_str); // Because the signature is in the URL it must be URL encoded.

1267

//printf(" Safe: \"%s\"\n", signed_str->getCString());

1268

push_(signed_str);

1269

1270

s3_buffer->setLength(0);

1271

s3_buffer->append("http://");

1272

s3_buffer->append(bucket);

1273

s3_buffer->append(".");

1274

s3_buffer->append(s3_server->getCString());

1275

s3_buffer->append(key);

1276

1277

s3_buffer->append("?AWSAccessKeyId=");

1278

s3_buffer->append(s3_public_key->getCString());

1279

s3_buffer->append("&Expires=");

1280

s3_buffer->append(timeout);

1281

s3_buffer->append("&Signature=");

1282

s3_buffer->append(signed_str->getCString());

1283

1284

release_(signed_str);

1285

1286

pop_(s3_buffer);

1287

CSString *str = CSString::newString(s3_buffer);

1288

return_(str);

1289

}

1290

1291

//#define S3_UNIT_TEST

1292

#ifdef S3_UNIT_TEST

1293

static void show_help_info(const char *cmd)

1294

{

1295

printf("Get authenticated query string:\n\t%s q <bucket> <object_key> <timeout>\n", cmd);

1296

printf("Delete object:\n\t%s d <bucket> <object_key>\n", cmd);

1297

printf("Delete all object with a given prefix:\n\t%s D <bucket> <object_prefix>\n", cmd);

1298

printf("Get object, data will be written to 'prottest.out':\n\t%s g <bucket> <object_key> <timeout>\n", cmd);

1299

printf("Get object header only:\n\t%s h <bucket> <object_key> <timeout>\n", cmd);

1300

printf("Put (Upload) an object:\n\t%s p <bucket> <object_key> <file>\n", cmd);

1301

printf("List objects in the bucket:\n\t%s l <bucket> [<object_prefix> [max_list_size]]\n", cmd);

1302

printf("Copy object:\n\t%s c <src_bucket> <src_object_key> <dst_bucket> <dst_object_key> \n", cmd);

1303

printf("Copy all object with a given prefix:\n\t%s C <src_bucket> <object_key_prefix> <dst_bucket> \n", cmd);

1304

}

1305

1306

void dump_headers(CSVector *header_array)

1307

{

1308

CSHTTPHeaders headers;

1309

1310

headers.setHeaders(header_array);

1311

printf("Reply Headers:\n");

1312

printf("--------------\n");

1313

1314

for (uint32_t i = 0; i < headers.numHeaders(); i++) {

1315

CSHeader *h = headers.getHeader(i);

1316

1317

printf("%s : %s\n", h->getNameCString(), h->getValueCString());

1318

h->release();

1319

}

1320

printf("--------------\n");

1321

headers.clearHeaders();

1322

}

1323

1324

int main(int argc, char **argv)

1325

{

1326

CSThread *main_thread;

1327

const char *pub_key;

1328

const char *priv_key;

1329

const char *server;

1330

CSS3Protocol *prot = NULL;

1331

1332

if (argc < 3) {

1333

show_help_info(argv[0]);

1334

return 0;

1335

}

1336

1337

if (! CSThread::startUp()) {

1338

CSException::throwException(CS_CONTEXT, ENOMEM, "CSThread::startUp() failed.");

1339

return 1;

1340

}

1341

1342

cs_init_memory();

1343

1344

main_thread = new CSThread( NULL);

1345

CSThread::setSelf(main_thread);

1346

1347

enter_();

1348

try_(a) {

1349

1350

pub_key = getenv("S3_ACCESS_KEY_ID");

1351

priv_key = getenv("S3_SECRET_ACCESS_KEY");

1352

new_(prot, CSS3Protocol());

1353

push_(prot);

1354

1355

server = getenv("S3_SERVER");

1356

if ((server == NULL) || (*server == 0))

1357

server = "s3.amazonaws.com/";

1358

prot->s3_setServer(server);

1359

prot->s3_setPublicKey(pub_key);

1360

prot->s3_setPrivateKey(priv_key);

1361

1362

switch (argv[1][0]) {

1363

case 'q': // Get the query string

1364

if (argc == 5) {

1365

CSString *qstr = prot->s3_getDataURL(argv[2], argv[3], atoi(argv[4]));

1366

printf("To test call:\ncurl -L -D - \"%s\"\n", qstr->getCString());

1367

qstr->release();

1368

} else

1369

printf("Bad command: q <bucket> <object_key> <timeout>\n");

1370

1371

break;

1372

case 'd': // Delete the object

1373

if (argc == 4) {

1374

printf("delete %s %s\n", argv[2], argv[3]);

1375

if (!prot->s3_delete(argv[2], argv[3]))

1376

printf("%s/%s could not be found.\n", argv[2], argv[3]);

1377

1378

} else

1379

printf("Bad command: d <bucket> <object_key>\n");

1380

1381

break;

1382

case 'D': // Delete objects like

1383

if (argc == 4) {

1384

CSVector *list;

1385

CSString *key;

1386

1387

list = prot->s3_list(argv[2], argv[3]);

1388

push_(list);

1389

while (key = (CSString*) list->take(0)) {

1390

printf("Deleting %s\n", key->getCString());

1391

prot->s3_delete(argv[2], key->getCString());

1392

key->release();

1393

}

1394

release_(list);

1395

1396

} else

1397

printf("Bad command: D <bucket> <object_key_prefix>\n");

1398

1399

break;

1400

case 'g': // Get the object

1401

if ((argc == 4) || (argc == 6)) {

1402

CSFile *output;

1403

CSVector *headers;

1404

bool found;

1405

S3RangeRec *range_ptr = NULL, range = {0,0};

1406

1407

if (argc == 6) {

1408

range.startByte = atoi(argv[4]);

1409

range.endByte = atoi(argv[5]);

1410

range_ptr = &range;

1411

}

1412

1413

output = CSFile::newFile("prottest.out");

1414

push_(output);

1415

output->open(CSFile::CREATE | CSFile::TRUNCATE);

1416

headers = prot->s3_receive(output->getOutputStream(), argv[2], argv[3], &found, range_ptr);

1417

if (!found)

1418

printf("%s/%s could not be found.\n", argv[2], argv[3]);

1419

1420

dump_headers(headers);

1421

1422

release_(output);

1423

} else

1424

printf("Bad command: g <bucket> <object_key>\n");

1425

1426

break;

1427

1428

case 'h': // Get the object header

1429

if (argc == 4) {

1430

CSVector *headers;

1431

bool found;

1432

S3RangeRec range = {0,0};

1433

1434

headers = prot->s3_receive(NULL, argv[2], argv[3], &found);

1435

if (!found)

1436

printf("%s/%s could not be found.\n", argv[2], argv[3]);

1437

1438

dump_headers(headers);

1439

1440

} else

1441

printf("Bad command: h <bucket> <object_key>\n");

1442

1443

break;

1444

1445

case 'p': // Put (Upload) the object

1446

if (argc == 5) {

1447

CSFile *input;

1448

Md5Digest digest;

1449

CSVector *headers;

1450

1451

input = CSFile::newFile(argv[4]);

1452

push_(input);

1453

input->open(CSFile::READONLY);

1454

input->md5Digest(&digest);

1455

headers = prot->s3_send(input->getInputStream(), argv[2], argv[3], input->myFilePath->getSize(), NULL, &digest);

1456

dump_headers(headers);

1457

release_(input);

1458

} else

1459

printf("Bad command: p <bucket> <object_key> <file> \n");

1460

1461

break;

1462

1463

case 'c': // Copy the object

1464

if (argc == 6) {

1465

prot->s3_copy(NULL, argv[4], argv[5], argv[2], argv[3]);

1466

} else

1467

printf("Bad command: c <src_bucket> <src_object_key> <dst_bucket> <dst_object_key>\n");

1468

1469

break;

1470

1471

case 'C': // Copy objects like

1472

if (argc == 5) {

1473

CSVector *list;

1474

CSString *key;

1475

1476

list = prot->s3_list(argv[2], argv[3]);

1477

push_(list);

1478

while (key = (CSString*) list->take(0)) {

1479

printf("Copying %s\n", key->getCString());

1480

prot->s3_copy(NULL, argv[4], key->getCString(), argv[2], key->getCString());

1481

key->release();

1482

}

1483

release_(list);

1484

1485

} else

1486

printf("Bad command: C <src_bucket> <object_key_prefix> <dst_bucket>\n");

1487

1488

break;

1489

case 'l': // List the object

1490

if ((argc == 3) || (argc == 4) || (argc == 5)) {

1491

uint32_t max = 0;

1492

char *prefix = NULL;

1493

CSVector *list;

1494

CSString *key;

1495

1496

if (argc > 3) {

1497

prefix = argv[3];

1498

if (!strlen(prefix))

1499

prefix = NULL;

1500

}

1501

1502

if (argc == 5)

1503

max = atol(argv[4]);

1504

1505

list = prot->s3_list(argv[2], prefix, max);

1506

push_(list);

1507

while (key = (CSString*) list->take(0)) {

1508

printf("%s\n", key->getCString());

1509

key->release();

1510

}

1511

release_(list);

1512

1513

} else

1514

printf("Bad command: l <bucket> [<object_prefix> [max_list_size]] \n");

1515

1516

break;

1517

default:

1518

printf("Unknown command.\n");

1519

show_help_info(argv[0]);

1520

}

1521

1522

release_(prot);

1523

}

1524

1525

catch_(a);

1526

self->logException();

1527

1528

cont_(a);

1529

1530

outer_()

1531

main_thread->release();

1532

cs_exit_memory();

1533

CSThread::shutDown();

1534

return 0;

1535

}

1536

1537

#endif

1538

1539

Older »